From 4d2364d35321ef227d32b603e5a7fc0f67ceaee1 Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Tue, 9 Jan 2024 11:36:48 +0400 Subject: [PATCH] Make settings validation periodically (#374) --- .../dshackle/startup/ConfiguredUpstreams.kt | 16 +++- .../dshackle/startup/UpstreamChangeEvent.kt | 10 +++ .../configure/BitcoinUpstreamCreator.kt | 7 +- .../configure/EthereumUpstreamCreator.kt | 3 +- .../configure/GenericUpstreamCreator.kt | 13 ++- .../startup/configure/UpstreamCreator.kt | 5 +- .../startup/configure/UpstreamFactory.kt | 11 ++- .../dshackle/upstream/Multistream.kt | 49 ++++++++--- .../upstream/generic/GenericUpstream.kt | 87 ++++++++++++------- 9 files changed, 136 insertions(+), 65 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index a9ba7751..d4367ad2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.foundation.ChainOptions +import io.emeraldpay.dshackle.startup.configure.UpstreamCreationData import io.emeraldpay.dshackle.startup.configure.UpstreamFactory import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder import org.slf4j.LoggerFactory @@ -54,11 +55,20 @@ open class ConfiguredUpstreams( log.error("Chain is unknown: ${up.chain}") return@forEach } - val upstream = upstreamFactory.createUpstream(chain.type, up, defaultOptions) - if (upstream != null) { + val upstreamCreationData = upstreamFactory.createUpstream(chain.type, up, defaultOptions) + if (upstreamCreationData != UpstreamCreationData.default()) { + val eventType = if (upstreamCreationData.isValid) { + UpstreamChangeEvent.ChangeType.ADDED + } else { + UpstreamChangeEvent.ChangeType.OBSERVED + } multistreamHolder.getUpstream(chain) .processUpstreamsEvents( - UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED), + UpstreamChangeEvent( + chain, + upstreamCreationData.upstream!!, + eventType, + ), ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt index 8e6b9275..0ffc84a9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/UpstreamChangeEvent.kt @@ -59,6 +59,16 @@ data class UpstreamChangeEvent( * Upstream is removed (it still doesn't mean it wouldn't return again after some reconfiguration) */ REMOVED, + + /** + * Upstream is removed but not stopped due to upstream fatal settings error (it could return some reconfiguration) + */ + FATAL_SETTINGS_ERROR_REMOVED, + + /** + * Upstream is observed for state updates (it could be added later) + */ + OBSERVED, } override fun setCaches(caches: Caches) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt index 4836420e..3591b678 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/BitcoinUpstreamCreator.kt @@ -10,7 +10,6 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.CallTargetsHolder import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.MergedHead -import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcHead import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinZMQHead @@ -38,13 +37,13 @@ class BitcoinUpstreamCreator( chain: Chain, options: ChainOptions.Options, chainConf: ChainsConfig.ChainConfig, - ): Upstream? { + ): UpstreamCreationData { val config = upstreamsConfig.cast(UpstreamsConfig.BitcoinConnection::class.java) val conn = config.connection!! val httpFactory = genericConnectorFactoryCreator.buildHttpFactory(conn.rpc) if (httpFactory == null) { log.warn("Upstream doesn't have API configuration") - return null + return UpstreamCreationData.default() } val directApi = httpFactory.create(config.id, chain) val esplora = conn.esplora?.let { endpoint -> @@ -74,6 +73,6 @@ class BitcoinUpstreamCreator( methods, esplora, chainConf, ) upstream.start() - return upstream + return UpstreamCreationData(upstream, true) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt index 39c0224d..e5c9de67 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/EthereumUpstreamCreator.kt @@ -6,7 +6,6 @@ import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.upstream.CallTargetsHolder -import io.emeraldpay.dshackle.upstream.Upstream import org.springframework.stereotype.Component @Component @@ -22,7 +21,7 @@ class EthereumUpstreamCreator( chain: Chain, options: ChainOptions.Options, chainConf: ChainsConfig.ChainConfig, - ): Upstream? { + ): UpstreamCreationData { var rating = 0 val connection = if (upstreamsConfig.connection is UpstreamsConfig.EthereumPosConnection) { val posConn = upstreamsConfig.cast(UpstreamsConfig.EthereumPosConnection::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt index 7ea522f4..c616a9fa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/GenericUpstreamCreator.kt @@ -8,7 +8,6 @@ import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.BlockValidator import io.emeraldpay.dshackle.upstream.CallTargetsHolder -import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice import io.emeraldpay.dshackle.upstream.generic.ChainSpecificRegistry import io.emeraldpay.dshackle.upstream.generic.GenericUpstream @@ -31,7 +30,7 @@ open class GenericUpstreamCreator( chain: Chain, options: ChainOptions.Options, chainConf: ChainsConfig.ChainConfig, - ): Upstream? { + ): UpstreamCreationData { return buildGenericUpstream( upstreamsConfig.nodeId, upstreamsConfig, @@ -51,10 +50,10 @@ open class GenericUpstreamCreator( options: ChainOptions.Options, chainConfig: ChainsConfig.ChainConfig, nodeRating: Int, - ): Upstream? { + ): UpstreamCreationData { if (config.connection == null) { log.warn("Upstream doesn't have connection configuration") - return null + return UpstreamCreationData.default() } val cs = ChainSpecificRegistry.resolve(chain) @@ -66,7 +65,7 @@ open class GenericUpstreamCreator( NoChoiceWithPriorityForkChoice(nodeRating, config.id!!), BlockValidator.ALWAYS_VALID, chainConfig, - ) ?: return null + ) ?: return UpstreamCreationData.default() val methods = buildMethods(config, chain) @@ -93,9 +92,9 @@ open class GenericUpstreamCreator( upstream.start() if (!upstream.isRunning) { log.debug("Upstream ${upstream.getId()} is not running, it can't be added") - return null + return UpstreamCreationData.default() } - return upstream + return UpstreamCreationData(upstream, upstream.isValid()) } private fun getHash(nodeId: Int?, obj: Any): Byte = diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt index 786b9a88..52932cf2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamCreator.kt @@ -7,7 +7,6 @@ import io.emeraldpay.dshackle.config.IndexConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.upstream.CallTargetsHolder -import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import org.slf4j.Logger @@ -23,7 +22,7 @@ abstract class UpstreamCreator( fun createUpstream( upstreamsConfig: UpstreamsConfig.Upstream<*>, defaultOptions: Map, - ): Upstream? { + ): UpstreamCreationData { val chain = Global.chainById(upstreamsConfig.chain) if (chain == Chain.UNSPECIFIED) { throw IllegalArgumentException("Chain is unknown: ${upstreamsConfig.chain}") @@ -42,7 +41,7 @@ abstract class UpstreamCreator( chain: Chain, options: ChainOptions.Options, chainConf: ChainsConfig.ChainConfig, - ): Upstream? + ): UpstreamCreationData protected fun buildMethods(config: UpstreamsConfig.Upstream<*>, chain: Chain): CallMethods { return if (config.methods != null || config.methodGroups != null) { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt index c7097156..465f244b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/configure/UpstreamFactory.kt @@ -7,6 +7,15 @@ import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.upstream.Upstream import org.springframework.stereotype.Component +data class UpstreamCreationData( + val upstream: Upstream?, + val isValid: Boolean, +) { + companion object { + fun default() = UpstreamCreationData(null, false) + } +} + @Component class UpstreamFactory( private val genericUpstreamCreator: GenericUpstreamCreator, @@ -18,7 +27,7 @@ class UpstreamFactory( type: BlockchainType, upstreamsConfig: UpstreamsConfig.Upstream<*>, defaultOptions: Map, - ): Upstream? { + ): UpstreamCreationData { return when (type) { BlockchainType.ETHEREUM -> ethereumUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions) BlockchainType.BITCOIN -> bitcoinUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 90a0faaa..e2b9aa2e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -84,6 +84,9 @@ abstract class Multistream( @Volatile private var quorumLabels: List? = null private val meters: MutableMap> = HashMap() + private val observedUpstreams = Sinks.many() + .multicast() + .directBestEffort() private val addedUpstreams = Sinks.many() .multicast() .directBestEffort() @@ -153,6 +156,15 @@ abstract class Multistream( .subscribe { onUpstreamChange(it) } + + observedUpstreams.asFlux() + .flatMap { + it.observeState() + .takeUntil { event -> event.type == UpstreamChangeEvent.ChangeType.ADDED } + } + .subscribe { + this.processUpstreamsEvents(it) + } } /** @@ -176,10 +188,10 @@ abstract class Multistream( } } - fun removeUpstream(id: String): Boolean = + fun removeUpstream(id: String, stopUpstream: Boolean): Boolean = getUpstreams().removeIf { up -> (up.getId() == id).also { - if (it) { + if (it && stopUpstream) { up.stop() } } @@ -332,7 +344,7 @@ abstract class Multistream( .takeUntilOther( subscribeRemovedUpstreams() .filter { - it.getId() == upstream.getId() + it.getId() == upstream.getId() && !it.isRunning() }, ) } @@ -460,16 +472,29 @@ abstract class Multistream( } UpstreamChangeEvent.ChangeType.REMOVED -> { - removeUpstream(event.upstream.getId()).takeIf { it }?.let { - try { - removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } - onUpstreamsUpdated() - log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed") - } catch (e: Sinks.EmissionException) { - log.error("error during event processing $event", e) - } - } + removeUpstream(event, true) } + + UpstreamChangeEvent.ChangeType.FATAL_SETTINGS_ERROR_REMOVED -> { + removeUpstream(event, false) + } + + UpstreamChangeEvent.ChangeType.OBSERVED -> { + observedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + log.info("Upstream ${event.upstream.getId()} with chain $chain has been added to the observation") + } + } + } + } + + private fun removeUpstream(event: UpstreamChangeEvent, stopUpstream: Boolean) { + removeUpstream(event.upstream.getId(), stopUpstream).takeIf { it }?.let { + try { + removedUpstreams.emitNext(event.upstream) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + onUpstreamsUpdated() + log.info("Upstream ${event.upstream.getId()} with chain $chain has been removed") + } catch (e: Sinks.EmissionException) { + log.error("error during event processing $event", e) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt index 997aafce..1d8cd1ed 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/generic/GenericUpstream.kt @@ -8,6 +8,7 @@ import io.emeraldpay.dshackle.foundation.ChainOptions import io.emeraldpay.dshackle.reader.JsonRpcReader import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.startup.UpstreamChangeEvent +import io.emeraldpay.dshackle.startup.UpstreamChangeEvent.ChangeType.UPDATED import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.Head @@ -57,6 +58,9 @@ open class GenericUpstream( private val lowerBoundBlockDetector = lowerBoundBlockDetectorBuilder(chain, this) + private val started = AtomicBoolean(false) + private val isUpstreamValid = AtomicBoolean(false) + override fun getHead(): Head { return connector.getHead() } @@ -108,44 +112,55 @@ open class GenericUpstream( return } ValidateUpstreamSettingsResult.UPSTREAM_SETTINGS_ERROR -> { - validateUpstreamSettings() + log.warn("Non fatal upstream settings error, continue validation...") } - else -> { + ValidateUpstreamSettingsResult.UPSTREAM_VALID -> { + isUpstreamValid.set(true) upstreamStart() } } + validateUpstreamSettings() } else { + isUpstreamValid.set(true) upstreamStart() } + + started.set(true) } private fun validateUpstreamSettings() { if (validator != null) { validationSettingsSubscription = Flux.interval( - Duration.ofSeconds(10), Duration.ofSeconds(20), ).flatMap { validator.validateUpstreamSettings() - }.subscribe { - when (it) { - ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR -> { - connector.stop() - disposeValidationSettingsSubscription() - } + } + .distinctUntilChanged() + .subscribe { + when (it) { + ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR -> { + if (isUpstreamValid.get()) { + log.warn("There is a fatal error after upstream settings validation, removing ${getId()}...") + partialStop() + sendUpstreamStateEvent(UpstreamChangeEvent.ChangeType.FATAL_SETTINGS_ERROR_REMOVED) + } + isUpstreamValid.set(false) + } - ValidateUpstreamSettingsResult.UPSTREAM_VALID -> { - upstreamStart() - stateEventStream.emitNext( - UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.ADDED), - ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } - disposeValidationSettingsSubscription() - } + ValidateUpstreamSettingsResult.UPSTREAM_VALID -> { + if (!isUpstreamValid.get()) { + log.warn("Upstream ${getId()} is now valid, adding to the multistream...") + upstreamStart() + sendUpstreamStateEvent(UpstreamChangeEvent.ChangeType.ADDED) + } + isUpstreamValid.set(true) + } - else -> { - log.warn("Continue validation of upstream ${getId()}") + else -> { + log.warn("Continue validation of upstream ${getId()}") + } } } - } } } @@ -165,9 +180,7 @@ open class GenericUpstream( } livenessSubscription = connector.hasLiveSubscriptionHead().subscribe({ hasLiveSubscriptionHead.set(it) - stateEventStream.emitNext( - UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.UPDATED), - ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + sendUpstreamStateEvent(UPDATED) }, { log.debug("Error while checking live subscription for ${getId()}", it) },) @@ -177,19 +190,21 @@ open class GenericUpstream( } override fun stop() { + partialStop() + validationSettingsSubscription?.dispose() + validationSettingsSubscription = null + connector.stop() + started.set(false) + } + + private fun partialStop() { validatorSubscription?.dispose() validatorSubscription = null livenessSubscription?.dispose() livenessSubscription = null lowerBlockDetectorSubscription?.dispose() lowerBlockDetectorSubscription = null - disposeValidationSettingsSubscription() - connector.stop() - } - - private fun disposeValidationSettingsSubscription() { - validationSettingsSubscription?.dispose() - validationSettingsSubscription = null + connector.getHead().stop() } private fun updateLabels(label: Pair) { @@ -202,9 +217,7 @@ open class GenericUpstream( private fun detectLowerBlock() { lowerBlockDetectorSubscription = lowerBoundBlockDetector.lowerBlock() .subscribe { - stateEventStream.emitNext( - UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.UPDATED), - ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + sendUpstreamStateEvent(UPDATED) } } @@ -212,5 +225,13 @@ open class GenericUpstream( return connector.getIngressSubscription() } - override fun isRunning() = connector.isRunning() && validationSettingsSubscription == null + override fun isRunning() = connector.isRunning() || started.get() + + fun isValid(): Boolean = isUpstreamValid.get() + + private fun sendUpstreamStateEvent(eventType: UpstreamChangeEvent.ChangeType) { + stateEventStream.emitNext( + UpstreamChangeEvent(chain, this, eventType), + ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED } + } }