Make settings validation periodically (#374)

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)
+                val upstreamCreationData = upstreamFactory.createUpstream(chain.type, up, defaultOptions)
-                if (upstream != null) {
+                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,
+                            ),
                         )
                 }
             }

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) {

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)
     }
 }

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)

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 =

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<Chain, ChainOptions.PartialOptions>,
-    ): 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) {

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<Chain, ChainOptions.PartialOptions>,
-    ): Upstream? {
+    ): UpstreamCreationData {
         return when (type) {
             BlockchainType.ETHEREUM -> ethereumUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions)
             BlockchainType.BITCOIN -> bitcoinUpstreamCreator.createUpstream(upstreamsConfig, defaultOptions)

src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt

@@ -84,6 +84,9 @@ abstract class Multistream(
     @Volatile
     private var quorumLabels: List<QuorumForLabels.QuorumItem>? = null
     private val meters: MutableMap<String, List<Meter.Id>> = HashMap()
+    private val observedUpstreams = Sinks.many()
+        .multicast()
+        .directBestEffort<Upstream>()
     private val addedUpstreams = Sinks.many()
         .multicast()
         .directBestEffort<Upstream>()
@@ -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 {
+                    removeUpstream(event, true)
-                        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)
-                        }
-                    }
                 }
+
+                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)
             }
         }
     }

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) {
+                .distinctUntilChanged()
-                    ValidateUpstreamSettingsResult.UPSTREAM_FATAL_SETTINGS_ERROR -> {
+                .subscribe {
-                        connector.stop()
+                    when (it) {
-                        disposeValidationSettingsSubscription()
+                        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 -> {
+                        ValidateUpstreamSettingsResult.UPSTREAM_VALID -> {
-                        upstreamStart()
+                            if (!isUpstreamValid.get()) {
-                        stateEventStream.emitNext(
+                                log.warn("Upstream ${getId()} is now valid, adding to the multistream...")
-                            UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.ADDED),
+                                upstreamStart()
-                        ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
+                                sendUpstreamStateEvent(UpstreamChangeEvent.ChangeType.ADDED)
-                        disposeValidationSettingsSubscription()
+                            }
-                    }
+                            isUpstreamValid.set(true)
+                        }
 
-                    else -> {
+                        else -> {
-                        log.warn("Continue validation of upstream ${getId()}")
+                            log.warn("Continue validation of upstream ${getId()}")
+                        }
                     }
                 }
-            }
         }
     }
 
@@ -165,9 +180,7 @@ open class GenericUpstream(
         }
         livenessSubscription = connector.hasLiveSubscriptionHead().subscribe({
             hasLiveSubscriptionHead.set(it)
-            stateEventStream.emitNext(
+            sendUpstreamStateEvent(UPDATED)
-                UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.UPDATED),
-            ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
         }, {
             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.getHead().stop()
-        connector.stop()
-    }
-
-    private fun disposeValidationSettingsSubscription() {
-        validationSettingsSubscription?.dispose()
-        validationSettingsSubscription = null
     }
 
     private fun updateLabels(label: Pair<String, String>) {
@@ -202,9 +217,7 @@ open class GenericUpstream(
     private fun detectLowerBlock() {
         lowerBlockDetectorSubscription = lowerBoundBlockDetector.lowerBlock()
             .subscribe {
-                stateEventStream.emitNext(
+                sendUpstreamStateEvent(UPDATED)
-                    UpstreamChangeEvent(chain, this, UpstreamChangeEvent.ChangeType.UPDATED),
-                ) { _, res -> res == Sinks.EmitResult.FAIL_NON_SERIALIZED }
             }
     }
 
@@ -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 }
+    }
 }