From 621d8671bec94ac2dd77dab174b655d2b4602c56 Mon Sep 17 00:00:00 2001 From: Vyacheslav Date: Thu, 10 Aug 2023 11:45:47 +0300 Subject: [PATCH] better track ws capabiltity for upstream through head liveness check (#277) --- .../dshackle/config/ChainsConfig.kt | 11 +++- .../dshackle/config/ChainsConfigReader.kt | 3 + .../dshackle/config/UpstreamsConfig.kt | 4 ++ .../dshackle/config/YamlConfigReader.kt | 12 ++++ .../dshackle/startup/ConfiguredUpstreams.kt | 18 ++++-- .../ethereum/EthereumLikeRpcUpstream.kt | 26 +++++---- .../ethereum/HeadLivenessValidator.kt | 41 +++++++++++++ .../ethereum/connectors/EthereumConnector.kt | 3 +- .../connectors/EthereumConnectorFactory.kt | 10 +++- .../connectors/EthereumRpcConnector.kt | 13 ++++- .../connectors/EthereumWsConnector.kt | 14 +++-- src/main/resources/chains.yaml | 22 +++++++ .../config/ChainsConfigReaderSpec.groovy | 6 ++ .../config/UpstreamsConfigReaderSpec.groovy | 2 +- .../dshackle/test/ConnectorFactoryMock.groovy | 8 +-- .../test/EthereumConnectorMock.groovy | 11 ++-- .../test/EthereumPosRpcUpstreamMock.groovy | 18 +++--- .../test/EthereumRpcUpstreamMock.groovy | 3 +- .../dshackle/test/TestingCommons.groovy | 5 -- .../dshackle/upstream/FilteredApisSpec.groovy | 6 +- .../EthereumEgressSubscriptionSpec.groovy | 13 ++++- .../ethereum/HeadLivenessValidatorSpec.groovy | 57 +++++++++++++++++++ src/test/resources/configs/chains-basic.yaml | 3 + 23 files changed, 249 insertions(+), 60 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidator.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidatorSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/ChainsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/ChainsConfig.kt index 93c75008..95ca85f6 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/ChainsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/ChainsConfig.kt @@ -1,6 +1,7 @@ package io.emeraldpay.dshackle.config import io.emeraldpay.dshackle.Chain +import java.time.Duration data class ChainsConfig(private val chains: Map, val currentDefault: RawChainConfig?) { companion object { @@ -9,6 +10,7 @@ data class ChainsConfig(private val chains: Map, val curr } data class RawChainConfig( + var expectedBlockTime: Duration? = null, var syncingLagSize: Int? = null, var laggingLagSize: Int? = null, var callLimitContract: String? = null, @@ -25,6 +27,7 @@ data class ChainsConfig(private val chains: Map, val curr } data class ChainConfig( + val expectedBlockTime: Duration, val syncingLagSize: Int, val laggingLagSize: Int, val options: UpstreamsConfig.PartialOptions, @@ -32,7 +35,7 @@ data class ChainsConfig(private val chains: Map, val curr ) { companion object { @JvmStatic - fun default() = ChainConfig(6, 1, UpstreamsConfig.PartialOptions(), null) + fun default() = ChainConfig(Duration.ofSeconds(12), 6, 1, UpstreamsConfig.PartialOptions(), null) } } @@ -45,7 +48,8 @@ data class ChainsConfig(private val chains: Map, val curr laggingLagSize = raw.laggingLagSize ?: default.laggingLagSize ?: panic(), syncingLagSize = raw.syncingLagSize ?: default.syncingLagSize ?: panic(), options = options, - callLimitContract = raw.callLimitContract + callLimitContract = raw.callLimitContract, + expectedBlockTime = raw.expectedBlockTime ?: default.expectedBlockTime ?: panic(), ) } @@ -61,7 +65,8 @@ data class ChainsConfig(private val chains: Map, val curr syncingLagSize = patch?.syncingLagSize ?: current.syncingLagSize, laggingLagSize = patch?.laggingLagSize ?: current.laggingLagSize, options = patch?.options ?: current.options, - callLimitContract = patch?.callLimitContract ?: current.callLimitContract + callLimitContract = patch?.callLimitContract ?: current.callLimitContract, + expectedBlockTime = patch?.expectedBlockTime ?: current.expectedBlockTime ) private fun merge( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/ChainsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/ChainsConfigReader.kt index df1258c5..3d2fc974 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/ChainsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/ChainsConfigReader.kt @@ -53,6 +53,9 @@ class ChainsConfigReader( rawConfig.laggingLagSize = it } } + getValueAsDuration(node, "expected-block-time")?.let { + rawConfig.expectedBlockTime = it + } getValueAsString(node, "call-validate-contract")?.let { rawConfig.callLimitContract = it } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 448eb606..171c1554 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -30,6 +30,7 @@ open class UpstreamsConfig { var upstreams: MutableList> = ArrayList>() data class Options( + val disableUpstreamValidation: Boolean, val disableValidation: Boolean, val validationInterval: Int, val timeout: Duration, @@ -43,6 +44,7 @@ open class UpstreamsConfig { open class PartialOptions { var disableValidation: Boolean? = null + var disableUpstreamValidation: Boolean? = null var validationInterval: Int? = null set(value) { require(value == null || value > 0) { @@ -78,11 +80,13 @@ open class UpstreamsConfig { copy.validateCalllimit = firstNonNull(overwrites.validateCalllimit, this.validateCalllimit) copy.timeout = firstNonNull(overwrites.timeout, this.timeout) copy.validateChain = firstNonNull(overwrites.validateChain, this.validateChain) + copy.disableUpstreamValidation = firstNonNull(overwrites.disableUpstreamValidation, this.disableUpstreamValidation) return copy } fun buildOptions(): Options = Options( + firstNonNull(this.disableUpstreamValidation, false)!!, firstNonNull(this.disableValidation, false)!!, firstNonNull(this.validationInterval, 30)!!, firstNonNull(this.timeout, Defaults.timeout)!!, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt index 9a277c0a..fb6f94e3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt @@ -24,6 +24,8 @@ import org.yaml.snakeyaml.nodes.ScalarNode import java.io.InputStream import java.io.InputStreamReader import java.util.Locale +import kotlin.time.Duration +import kotlin.time.toJavaDuration abstract class YamlConfigReader : ConfigReader { private val envVariables = EnvVariables() @@ -92,6 +94,16 @@ abstract class YamlConfigReader : ConfigReader { }?.let(envVariables::postProcess) } + protected fun getValueAsDuration(mappingNode: MappingNode?, key: String): java.time.Duration? { + return getValue(mappingNode, key)?.let { + return@let if (it.isPlain) { + Duration.parse(it.value).toJavaDuration() + } else { + null + } + } + } + protected fun getValueAsInt(mappingNode: MappingNode?, key: String): Int? { return getValue(mappingNode, key)?.let { return@let if (it.isPlain) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 74bafc1e..a71d53fe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -215,7 +215,8 @@ open class ConfiguredUpstreams( chain, urls, NoChoiceWithPriorityForkChoice(conn.upstreamRating, config.id!!), - BlockValidator.ALWAYS_VALID + BlockValidator.ALWAYS_VALID, + chainConf ) val methods = buildMethods(config, chain) if (connectorFactory == null) { @@ -236,7 +237,8 @@ open class ConfiguredUpstreams( QuorumForLabels.QuorumItem(1, config.labels), connectorFactory, chainConf, - true + true, + eventPublisher ) upstream.start() if (!upstream.isRunning) return null @@ -304,7 +306,8 @@ open class ConfiguredUpstreams( chain, urls, MostWorkForkChoice(), - EthereumBlockValidator() + EthereumBlockValidator(), + chainConf ) if (connectorFactory == null) { return null @@ -320,7 +323,8 @@ open class ConfiguredUpstreams( QuorumForLabels.QuorumItem(1, config.labels), connectorFactory, chainConf, - false + false, + eventPublisher ) upstream.start() return upstream @@ -413,7 +417,8 @@ open class ConfiguredUpstreams( chain: Chain, urls: ArrayList, forkChoice: ForkChoice, - blockValidator: BlockValidator + blockValidator: BlockValidator, + chainsConf: ChainsConfig.ChainConfig ): EthereumConnectorFactory? { val wsFactoryApi = buildWsFactory(id, chain, conn, urls) val httpFactory = buildHttpFactory(conn, urls) @@ -426,7 +431,8 @@ open class ConfiguredUpstreams( forkChoice, blockValidator, wsConnectionResubscribeScheduler, - headScheduler + headScheduler, + chainsConf.expectedBlockTime ) if (!connectorFactory.isValid()) { log.warn("Upstream configuration is invalid (probably no http endpoint)") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt index e4e12c8c..0ab1056f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt @@ -23,6 +23,7 @@ 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.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream @@ -30,10 +31,11 @@ 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 io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumLabelsDetector +import org.springframework.context.ApplicationEventPublisher import org.springframework.context.Lifecycle import reactor.core.Disposable +import java.util.concurrent.atomic.AtomicBoolean open class EthereumLikeRpcUpstream( id: String, @@ -45,21 +47,21 @@ open class EthereumLikeRpcUpstream( private val node: QuorumForLabels.QuorumItem?, connectorFactory: ConnectorFactory, chainConfig: ChainsConfig.ChainConfig, - skipEnhance: Boolean + skipEnhance: Boolean, + private val eventPublisher: ApplicationEventPublisher? ) : EthereumLikeUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled { private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(chain, this, getOptions(), chainConfig.callLimitContract) - private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, skipEnhance) + protected val connector: EthereumConnector = connectorFactory.create(this, validator, chain, skipEnhance) private val labelsDetector = EthereumLabelsDetector(this.getIngressReader()) + private var hasLiveSubscriptionHead: AtomicBoolean = AtomicBoolean(false) private var validatorSubscription: Disposable? = null override fun getCapabilities(): Set { - return when (connector.getConnectorMode()) { - EthereumConnectorFactory.ConnectorMode.WS_ONLY, - EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD, - EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD -> - setOf(Capability.RPC, Capability.BALANCE, Capability.WS_HEAD) - EthereumConnectorFactory.ConnectorMode.RPC_ONLY -> setOf(Capability.RPC, Capability.BALANCE) + return if (hasLiveSubscriptionHead.get()) { + setOf(Capability.RPC, Capability.BALANCE, Capability.WS_HEAD) + } else { + setOf(Capability.RPC, Capability.BALANCE) } } @@ -72,7 +74,7 @@ open class EthereumLikeRpcUpstream( override fun start() { log.info("Configured for ${chain.chainName}") connector.start() - if (!validator.validateUpstreamSettings()) { + if (!getOptions().disableUpstreamValidation && !validator.validateUpstreamSettings()) { connector.stop() log.warn("Upstream ${getId()} couldn't start, invalid upstream settings") return @@ -86,6 +88,10 @@ open class EthereumLikeRpcUpstream( validatorSubscription = validator.start() .subscribe(this::setStatus) } + connector.hasLiveSubscriptionHead().subscribe { + hasLiveSubscriptionHead.set(it) + eventPublisher?.publishEvent(UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.UPDATED)) + } labelsDetector.detectLabels() .toStream() .forEach { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidator.kt new file mode 100644 index 00000000..6f86a140 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidator.kt @@ -0,0 +1,41 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.upstream.Head +import reactor.core.publisher.Flux +import reactor.core.scheduler.Scheduler +import java.time.Duration + +class HeadLivenessValidator( + val head: Head, + val expectedBlockTime: Duration, + val scheduler: Scheduler +) { + + companion object { + const val CHECKED_BLOCKS_UNTIL_LIVE = 3 + } + + fun getFlux(): Flux { + // first we have moving window of 2 blocks and check that they are consecutive ones + return head.getFlux().buffer(2, 1).map { + it.last().height - it.first().height == 1L + }.scan(Pair(0, true)) { acc, value -> + // then we accumulate consecutive true events, false resets counter + if (value) { + Pair(acc.first + 1, true) + } else { + Pair(0, false) + } + }.flatMap { (count, value) -> + // we emit when we have false or checked CHECKED_BLOCKS_UNTIL_LIVE blocks + // CHECKED_BLOCKS_UNTIL_LIVE blocks == (CHECKED_BLOCKS_UNTIL_LIVE - 1) consecutive true + when { + count == (CHECKED_BLOCKS_UNTIL_LIVE - 1) -> Flux.just(true) + !value -> Flux.just(false) + else -> Flux.empty() + } + // finally, we timeout after we waited for double the time we needed to emit those blocks + }.timeout(expectedBlockTime.multipliedBy(CHECKED_BLOCKS_UNTIL_LIVE.toLong() * 2), Flux.just(false)) + .distinctUntilChanged().subscribeOn(scheduler) + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt index 0a100cb3..4207ad6a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnector.kt @@ -4,11 +4,12 @@ import io.emeraldpay.dshackle.reader.JsonRpcReader import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription +import reactor.core.publisher.Flux interface EthereumConnector : Lifecycle { fun getHead(): Head - fun getConnectorMode(): EthereumConnectorFactory.ConnectorMode + fun hasLiveSubscriptionHead(): Flux fun getIngressReader(): JsonRpcReader diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt index a03e2e0e..a25fe496 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt @@ -12,6 +12,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFact import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.WS_ONLY import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import reactor.core.scheduler.Scheduler +import java.time.Duration open class EthereumConnectorFactory( private val connectorType: ConnectorMode, @@ -20,7 +21,8 @@ open class EthereumConnectorFactory( private val forkChoice: ForkChoice, private val blockValidator: BlockValidator, private val wsConnectionResubscribeScheduler: Scheduler, - private val headScheduler: Scheduler + private val headScheduler: Scheduler, + private val expectedBlockTime: Duration ) : ConnectorFactory { override fun isValid(): Boolean { @@ -57,7 +59,8 @@ open class EthereumConnectorFactory( blockValidator, skipEnhance, wsConnectionResubscribeScheduler, - headScheduler + headScheduler, + expectedBlockTime ) } if (httpFactory == null) { @@ -72,7 +75,8 @@ open class EthereumConnectorFactory( blockValidator, skipEnhance, wsConnectionResubscribeScheduler, - headScheduler + headScheduler, + expectedBlockTime ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt index 6925c000..f0c82390 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt @@ -11,6 +11,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead +import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidator import io.emeraldpay.dshackle.upstream.ethereum.NoEthereumIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptionsImpl @@ -22,11 +23,12 @@ import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFact import io.emeraldpay.dshackle.upstream.forkchoice.AlwaysForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux import reactor.core.scheduler.Scheduler import java.time.Duration class EthereumRpcConnector( - private val connectorType: ConnectorMode, + connectorType: ConnectorMode, private val directReader: JsonRpcReader, wsFactory: EthereumWsConnectionPoolFactory?, id: String, @@ -34,16 +36,20 @@ class EthereumRpcConnector( blockValidator: BlockValidator, skipEnhance: Boolean, wsConnectionResubscribeScheduler: Scheduler, - headScheduler: Scheduler + headScheduler: Scheduler, + expectedBlockTime: Duration ) : EthereumConnector, CachesEnabled { private val pool: WsConnectionPool? private val head: Head + private val liveness: HeadLivenessValidator companion object { private val log = LoggerFactory.getLogger(EthereumRpcConnector::class.java) } - override fun getConnectorMode() = connectorType + override fun hasLiveSubscriptionHead(): Flux { + return liveness.getFlux() + } init { pool = wsFactory?.create(null) @@ -93,6 +99,7 @@ class EthereumRpcConnector( ) } } + liveness = HeadLivenessValidator(head, expectedBlockTime, headScheduler) } override fun setCaches(caches: Caches) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt index ffe4e61b..ae79bd2a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt @@ -7,12 +7,15 @@ import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead +import io.emeraldpay.dshackle.upstream.ethereum.HeadLivenessValidator import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptionsImpl import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsIngressSubscription import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient +import reactor.core.publisher.Flux import reactor.core.scheduler.Scheduler +import java.time.Duration class EthereumWsConnector( wsFactory: EthereumWsConnectionPoolFactory, @@ -21,13 +24,14 @@ class EthereumWsConnector( blockValidator: BlockValidator, skipEnhance: Boolean, wsConnectionResubscribeScheduler: Scheduler, - headScheduler: Scheduler + headScheduler: Scheduler, + expectedBlockTime: Duration ) : EthereumConnector { private val pool: WsConnectionPool private val reader: JsonRpcReader private val head: EthereumWsHead private val subscriptions: EthereumIngressSubscription - + private val liveness: HeadLivenessValidator init { pool = wsFactory.create(upstream) reader = JsonRpcWsClient(pool) @@ -42,11 +46,13 @@ class EthereumWsConnector( wsConnectionResubscribeScheduler, headScheduler ) + liveness = HeadLivenessValidator(head, expectedBlockTime, headScheduler) subscriptions = EthereumWsIngressSubscription(wsSubscriptions) } - override fun getConnectorMode() = EthereumConnectorFactory.ConnectorMode.WS_ONLY - + override fun hasLiveSubscriptionHead(): Flux { + return liveness.getFlux() + } override fun start() { pool.connect() head.start() diff --git a/src/main/resources/chains.yaml b/src/main/resources/chains.yaml index fd2dcacc..f4abbcbc 100644 --- a/src/main/resources/chains.yaml +++ b/src/main/resources/chains.yaml @@ -2,95 +2,112 @@ version: v1 chain-settings: default: + expected-block-time: 12s lags: syncing: 6 lagging: 1 chains: - id: eth call-validate-contract: 0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96 + expected-block-time: 12s lags: syncing: 6 lagging: 1 - id: goerli call-validate-contract: 0xCD9303A1F6da2a68f465A579a24cc2Ee5AE2192f + expected-block-time: 12s lags: syncing: 6 lagging: 1 - id: polygon + expected-block-time: 2.7s lags: syncing: 20 lagging: 10 - id: polygon-mumbai + expected-block-time: 2.7s lags: syncing: 20 lagging: 10 - id: arbitrum + expected-block-time: 260ms options: validate-peers: false lags: syncing: 40 lagging: 20 - id: arbitrum-testnet + expected-block-time: 1s options: validate-peers: false lags: syncing: 40 lagging: 20 - id: optimism + expected-block-time: 2s options: validate-peers: false lags: syncing: 40 lagging: 20 - id: optimism-testnet + expected-block-time: 2s options: validate-peers: false lags: syncing: 40 lagging: 20 - id: arbitrum-nova + expected-block-time: 1s options: disable-validation: true lags: syncing: 40 lagging: 20 - id: polygon-zkevm + expected-block-time: 2.7s options: disable-validation: true lags: syncing: 40 lagging: 20 - id: polygon-zkevm-testnet + expected-block-time: 1m options: disable-validation: true lags: syncing: 40 lagging: 20 - id: zksync + expected-block-time: 5s options: disable-validation: true lags: syncing: 40 lagging: 20 - id: zksync-testnet + expected-block-time: 5s options: disable-validation: true lags: syncing: 40 lagging: 20 - id: base + expected-block-time: 2s options: validate-peers: false lags: syncing: 40 lagging: 20 - id: base-goerli + expected-block-time: 2s options: validate-peers: false lags: syncing: 40 lagging: 20 - id: avalanche + expected-block-time: 2s options: validate-peers: false validate-syncing: false @@ -98,6 +115,7 @@ chain-settings: syncing: 10 lagging: 5 - id: avalanche-fuji + expected-block-time: 2s options: validate-peers: false validate-syncing: false @@ -105,24 +123,28 @@ chain-settings: syncing: 10 lagging: 5 - id: gnosis + expected-block-time: 6s options: validate-peers: false lags: syncing: 10 lagging: 5 - id: gnosis-chiado + expected-block-time: 6s options: validate-peers: false lags: syncing: 10 lagging: 5 - id: fantom + expected-block-time: 3s options: validate-peers: false lags: syncing: 10 lagging: 5 - id: fantom-testnet + expected-block-time: 1m options: validate-peers: false lags: diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/ChainsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/ChainsConfigReaderSpec.groovy index c74e1441..7a368a34 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/ChainsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/ChainsConfigReaderSpec.groovy @@ -20,6 +20,8 @@ import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.FileResolver import spock.lang.Specification +import java.time.Duration + class ChainsConfigReaderSpec extends Specification { ChainsConfigReader reader = new ChainsConfigReader( @@ -39,15 +41,19 @@ class ChainsConfigReaderSpec extends Specification { eth.laggingLagSize == 1 eth.syncingLagSize == 6 eth.callLimitContract == "0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96" + eth.expectedBlockTime == Duration.ofSeconds(12) pol.laggingLagSize == 10 pol.syncingLagSize == 20 + pol.expectedBlockTime == Duration.ofMillis(2700) opt.laggingLagSize == 3 opt.syncingLagSize == 40 opt.options.validatePeers == false + opt.expectedBlockTime == Duration.ofMillis(400) sep.laggingLagSize == 1 sep.syncingLagSize == 10 + sep.expectedBlockTime == Duration.ofSeconds(12) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 5308f510..a411640a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -635,7 +635,7 @@ class UpstreamsConfigReaderSpec extends Specification { def options = partialOptions.buildOptions() then: options == new UpstreamsConfig.Options( - false, 30, Duration.ofSeconds(60), null, true, 1, true, true, true + false, false, 30, Duration.ofSeconds(60), null, true, 1, true, true, true ) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy index b0c3ee98..6deba825 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy @@ -15,16 +15,10 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse class ConnectorFactoryMock implements ConnectorFactory { Reader api Head head - ConnectorMode mode ConnectorFactoryMock(Reader api, Head head) { - this(api, head, ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD) - } - - ConnectorFactoryMock(Reader api, Head head, ConnectorMode mode) { this.api = api this.head = head - this.mode = mode } boolean isValid() { @@ -32,6 +26,6 @@ class ConnectorFactoryMock implements ConnectorFactory { } EthereumConnector create(DefaultUpstream upstream, EthereumUpstreamValidator validator, Chain chain, boolean skipEnhance) { - return new EthereumConnectorMock(api, head, this.mode) + return new EthereumConnectorMock(api, head) } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy index 0497b927..9c3ad3a7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumConnectorMock.groovy @@ -9,21 +9,22 @@ import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFact import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import reactor.core.publisher.Flux class EthereumConnectorMock implements EthereumConnector { Reader api Head head - ConnectorMode mode + Flux liveness - EthereumConnectorMock(Reader api, Head head, ConnectorMode mode) { + EthereumConnectorMock(Reader api, Head head) { this.api = api - this.mode = mode this.head = head + this.liveness = Flux.just(false) } @Override - ConnectorMode getConnectorMode() { - return this.mode + Flux hasLiveSubscriptionHead() { + return liveness } @Override diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy index d00aba57..4f697574 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy @@ -27,7 +27,6 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.calls.* import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream -import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.jetbrains.annotations.NotNull @@ -36,7 +35,6 @@ import org.reactivestreams.Publisher class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream { EthereumHeadMock ethereumHeadMock - static CallMethods allMethods() { new AggregatedCallMethods([ new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET), @@ -50,7 +48,7 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream { } EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, Map labels) { - this(id, chain, api, allMethods(), labels, ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD) + this(id, chain, api, allMethods(), labels) } EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api) { @@ -62,18 +60,19 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream { } EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods) { - this(id, chain, api, methods, Collections.emptyMap(), ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD) + this(id, chain, api, methods, Collections.emptyMap()) } - EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods, Map labels, ConnectorMode mode) { + EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader api, CallMethods methods, Map labels) { super(id, (byte)id.hashCode(), chain, getOpts(), UpstreamsConfig.UpstreamRole.PRIMARY, methods, new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)), - new ConnectorFactoryMock(api, new EthereumHeadMock(), mode), + new ConnectorFactoryMock(api, new EthereumHeadMock()), ChainConfig.default(), - true + true, + null ) this.ethereumHeadMock = this.getHead() as EthereumHeadMock setLag(0) @@ -84,6 +83,7 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream { static Options getOpts() { def opt = UpstreamsConfig.PartialOptions.getDefaults() opt.setDisableValidation(true) + opt.setDisableUpstreamValidation(true) return opt.buildOptions() } @@ -91,6 +91,10 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream { this.ethereumHeadMock.nextBlock(block) } + EthereumConnectorMock getConnectorMock() { + return this.connector as EthereumConnectorMock + } + void setBlocks(Publisher blocks) { this.ethereumHeadMock.predefined = blocks } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy index da8c4c77..89e0f3f3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy @@ -62,7 +62,8 @@ class EthereumRpcUpstreamMock extends EthereumLikeRpcUpstream { new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()), new ConnectorFactoryMock(api, new EthereumHeadMock()), ChainsConfig.ChainConfig.default(), - false + false, + null ) this.ethereumHeadMock = this.getHead() as EthereumHeadMock setLag(0) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy index f33dc330..9619b918 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy @@ -30,7 +30,6 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream -import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.domain.BlockHash @@ -63,10 +62,6 @@ class TestingCommons { return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM__MAINNET, api()) } - static EthereumPosRpcUpstreamMock upstream(String id, ConnectorMode mode) { - return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM__MAINNET, api(), EthereumPosRpcUpstreamMock.allMethods(), Collections.emptyMap(), mode) - } - static EthereumPosRpcUpstreamMock upstream(String id, String provider) { return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM__MAINNET, api(), Collections.singletonMap("provider", provider)) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 397b827d..299429dd 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -59,7 +59,8 @@ class FilteredApisSpec extends Specification { new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID, Schedulers.boundedElastic(), - Schedulers.boundedElastic() + Schedulers.boundedElastic(), + Duration.ofSeconds(12) ) new EthereumLikeRpcUpstream( "test", @@ -71,7 +72,8 @@ class FilteredApisSpec extends Specification { new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)), connectorFactory, ChainsConfig.ChainConfig.default(), - false + false, + null ) } def matcher = new Selector.LabelMatcher("test", ["foo"]) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy index 9dfca957..6452a47d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumEgressSubscriptionSpec.groovy @@ -22,6 +22,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFact import io.emeraldpay.dshackle.upstream.ethereum.subscribe.PendingTxesSource import io.emeraldpay.etherjar.domain.Address import io.emeraldpay.etherjar.hex.Hex32 +import reactor.core.publisher.Flux import reactor.core.scheduler.Schedulers import spock.lang.Specification @@ -178,18 +179,26 @@ class EthereumEgressSubscriptionSpec extends Specification { def "get available subscriptions"() { when: - def up1 = TestingCommons.upstream("test", EthereumConnectorFactory.ConnectorMode.RPC_ONLY) + def up1 = TestingCommons.upstream("test") + up1.getConnectorMock().setLiveness(Flux.just(false)) + def ethereumSubscribe1 = new EthereumEgressSubscription(TestingCommons.multistream(up1) as EthereumPosMultiStream, Schedulers.boundedElastic(), null) then: ethereumSubscribe1.getAvailableTopics() == [] when: def up2 = TestingCommons.upstream("test") + up2.getConnectorMock().setLiveness(Flux.just(true)) + up2.stop() + up2.start() def ethereumSubscribe2 = new EthereumEgressSubscription(TestingCommons.multistream(up2) as EthereumPosMultiStream, Schedulers.boundedElastic(), null) then: ethereumSubscribe2.getAvailableTopics().toSet() == [EthereumEgressSubscription.METHOD_LOGS, EthereumEgressSubscription.METHOD_NEW_HEADS].toSet() when: def up3 = TestingCommons.upstream("test") - def ethereumSubscribe3 = new EthereumEgressSubscription(TestingCommons.multistream(up2) as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) + up3.getConnectorMock().setLiveness(Flux.just(true)) + up3.stop() + up3.start() + def ethereumSubscribe3 = new EthereumEgressSubscription(TestingCommons.multistream(up3) as EthereumPosMultiStream, Schedulers.boundedElastic(), Stub(PendingTxesSource)) then: ethereumSubscribe3.getAvailableTopics().toSet() == [EthereumEgressSubscription.METHOD_LOGS, EthereumEgressSubscription.METHOD_NEW_HEADS, EthereumEgressSubscription.METHOD_PENDING_TXES].toSet() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidatorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidatorSpec.groovy new file mode 100644 index 00000000..811c1f54 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/HeadLivenessValidatorSpec.groovy @@ -0,0 +1,57 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.test.EthereumHeadMock +import io.emeraldpay.dshackle.test.TestingCommons +import reactor.core.scheduler.Schedulers +import reactor.test.StepVerifier +import spock.lang.Specification + +import java.time.Duration + +class HeadLivenessValidatorSpec extends Specification{ + def "emits true"() { + when: + def head = new EthereumHeadMock() + def checker = new HeadLivenessValidator(head, Duration.ofSeconds(10), Schedulers.boundedElastic()) + then: + StepVerifier.create(checker.flux) + .then { + head.nextBlock(TestingCommons.blockForEthereum(1)) + head.nextBlock(TestingCommons.blockForEthereum(2)) + head.nextBlock(TestingCommons.blockForEthereum(3)) + }.expectNext(true).thenCancel().verify(Duration.ofSeconds(1)) + } + + def "starts accumulating trues but immediately emits after false"() { + when: + def head = new EthereumHeadMock() + def checker = new HeadLivenessValidator(head, Duration.ofSeconds(100), Schedulers.boundedElastic()) + then: + StepVerifier.create(checker.flux) + .then { + head.nextBlock(TestingCommons.blockForEthereum(1)) + head.nextBlock(TestingCommons.blockForEthereum(2)) + } + .expectNoEvent(Duration.ofMillis(100)) + .then { + head.nextBlock(TestingCommons.blockForEthereum(5)) + } + .expectNext(false) + .thenCancel().verify(Duration.ofSeconds(1)) + } + + def "starts accumulating trues but timeouts because head staled"() { + when: + def head = new EthereumHeadMock() + def checker = new HeadLivenessValidator(head, Duration.ofMillis(100), Schedulers.boundedElastic()) + then: + StepVerifier.create(checker.flux) + .then { + head.nextBlock(TestingCommons.blockForEthereum(1)) + head.nextBlock(TestingCommons.blockForEthereum(2)) + } + .thenAwait(Duration.ofSeconds(1)) + .expectNext(false) + .thenCancel().verify(Duration.ofSeconds(2)) + } +} diff --git a/src/test/resources/configs/chains-basic.yaml b/src/test/resources/configs/chains-basic.yaml index ac150e61..f5da51ac 100644 --- a/src/test/resources/configs/chains-basic.yaml +++ b/src/test/resources/configs/chains-basic.yaml @@ -5,13 +5,16 @@ chain-settings: lags: syncing: 6 lagging: 1 + expected-block-time: 12s chains: - id: eth call-validate-contract: 0x32268860cAAc2948Ab5DdC7b20db5a420467Cf96 + expected-block-time: 12s lags: syncing: 6 lagging: 1 - id: optimism + expected-block-time: 400ms lags: lagging: 3 - id: sepolia