better track ws capabiltity for upstream through head liveness check (#277)

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<Chain, RawChainConfig>, val currentDefault: RawChainConfig?) {
     companion object {
@@ -9,6 +10,7 @@ data class ChainsConfig(private val chains: Map<Chain, RawChainConfig>, 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<Chain, RawChainConfig>, 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<Chain, RawChainConfig>, 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<Chain, RawChainConfig>, 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<Chain, RawChainConfig>, 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(

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
         }

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt

@@ -30,6 +30,7 @@ open class UpstreamsConfig {
     var upstreams: MutableList<Upstream<*>> = ArrayList<Upstream<*>>()
 
     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)!!,

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<T> : ConfigReader<T> {
     private val envVariables = EnvVariables()
@@ -92,6 +94,16 @@ abstract class YamlConfigReader<T> : ConfigReader<T> {
         }?.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) {

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<URI>,
         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)")

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<Capability> {
-        return when (connector.getConnectorMode()) {
+        return if (hasLiveSubscriptionHead.get()) {
-            EthereumConnectorFactory.ConnectorMode.WS_ONLY,
+            setOf(Capability.RPC, Capability.BALANCE, Capability.WS_HEAD)
-            EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD,
+        } else {
-            EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD ->
+            setOf(Capability.RPC, Capability.BALANCE)
-                setOf(Capability.RPC, Capability.BALANCE, Capability.WS_HEAD)
-            EthereumConnectorFactory.ConnectorMode.RPC_ONLY -> 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 {

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<Boolean> {
+        // 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)
+    }
+}

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<Boolean>
 
     fun getIngressReader(): JsonRpcReader
 

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

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<Boolean> {
+        return liveness.getFlux()
+    }
 
     init {
         pool = wsFactory?.create(null)
@@ -93,6 +99,7 @@ class EthereumRpcConnector(
                 )
             }
         }
+        liveness = HeadLivenessValidator(head, expectedBlockTime, headScheduler)
     }
 
     override fun setCaches(caches: Caches) {

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<Boolean> {
-
+        return liveness.getFlux()
+    }
     override fun start() {
         pool.connect()
         head.start()

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:

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

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

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<JsonRpcRequest, JsonRpcResponse> api
     Head head
-    ConnectorMode mode
 
     ConnectorFactoryMock(Reader<JsonRpcRequest, JsonRpcResponse> api, Head head) {
-        this(api, head, ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD)
-    }
-
-    ConnectorFactoryMock(Reader<JsonRpcRequest, JsonRpcResponse> 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)
     }
 }

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<JsonRpcRequest, JsonRpcResponse> api
     Head head
-    ConnectorMode mode
+    Flux<Boolean> liveness
 
-    EthereumConnectorMock(Reader<JsonRpcRequest, JsonRpcResponse> api, Head head, ConnectorMode mode) {
+    EthereumConnectorMock(Reader<JsonRpcRequest, JsonRpcResponse> api, Head head) {
         this.api = api
-        this.mode = mode
         this.head = head
+        this.liveness = Flux.just(false)
     }
 
     @Override
-    ConnectorMode getConnectorMode() {
+    Flux<Boolean> hasLiveSubscriptionHead() {
-        return this.mode
+        return liveness
     }
 
     @Override

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<JsonRpcRequest, JsonRpcResponse> api, Map<String, String> 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<JsonRpcRequest, JsonRpcResponse> api) {
@@ -62,18 +60,19 @@ class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream {
     }
 
     EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {
-        this(id, chain, api, methods, Collections.<String, String>emptyMap(), ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD)
+        this(id, chain, api, methods, Collections.<String, String>emptyMap())
     }
 
-    EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods, Map<String, String> labels, ConnectorMode mode) {
+    EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods, Map<String, String> 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<BlockContainer> blocks) {
         this.ethereumHeadMock.predefined = blocks
     }

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)

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.<String, String>emptyMap(), mode)
-    }
-
     static EthereumPosRpcUpstreamMock upstream(String id, String provider) {
         return new EthereumPosRpcUpstreamMock(id, Chain.ETHEREUM__MAINNET, api(), Collections.singletonMap("provider", provider))
     }

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"])

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

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

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