Ignore blocks from syncing nodes (#243)

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -42,8 +42,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.ZMQServer
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumBlockValidator
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionFactory
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory
@@ -227,7 +226,7 @@ open class ConfiguredUpstreams(
             if (it.connectorMode == RPC_REQUESTS_WITH_MIXED_HEAD.name) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url
         }
         val hash = getHash(nodeId, hashUrl!!)
-        val upstream = EthereumPosRpcUpstream(
+        val upstream = EthereumLikeRpcUpstream(
             config.id!!,
             hash,
             chain,
@@ -236,7 +235,8 @@ open class ConfiguredUpstreams(
             methods,
             QuorumForLabels.QuorumItem(1, config.labels),
             connectorFactory,
-            chainConf
+            chainConf,
+            true
         )
         upstream.start()
         return upstream
@@ -291,7 +291,7 @@ open class ConfiguredUpstreams(
         chain: Chain,
         options: UpstreamsConfig.Options,
         chainConf: ChainsConfig.ChainConfig
-    ): EthereumRpcUpstream? {
+    ): Upstream? {
         val conn = config.connection!!
 
         val urls = ArrayList<URI>()
@@ -310,7 +310,7 @@ open class ConfiguredUpstreams(
         }
 
         val hashUrl = if (conn.connectorMode == RPC_REQUESTS_WITH_MIXED_HEAD.name) conn.rpc?.url ?: conn.ws?.url else conn.ws?.url ?: conn.rpc?.url
-        val upstream = EthereumRpcUpstream(
+        val upstream = EthereumLikeRpcUpstream(
             config.id!!,
             getHash(nodeId, hashUrl!!),
             chain,
@@ -318,7 +318,8 @@ open class ConfiguredUpstreams(
             methods,
             QuorumForLabels.QuorumItem(1, config.labels),
             connectorFactory,
-            chainConf
+            chainConf,
+            false
         )
         upstream.start()
         return upstream

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

@@ -149,7 +149,12 @@ abstract class AbstractHead @JvmOverloads constructor(
         future = null
     }
 
-    override fun onNoHeadUpdates() {
+    protected open fun onNoHeadUpdates() {
+        // NOOP
+    }
+
+    override fun onSyncingNode(isSyncing: Boolean) {
+        // NOOP
     }
 
     override fun start() {

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

@@ -37,6 +37,6 @@ class EmptyHead : Head {
     override fun stop() {
     }
 
-    override fun onNoHeadUpdates() {
+    override fun onSyncingNode(isSyncing: Boolean) {
     }
 }

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

@@ -42,5 +42,5 @@ interface Head {
 
     fun stop()
 
-    fun onNoHeadUpdates()
+    fun onSyncingNode(isSyncing: Boolean)
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt

@@ -49,7 +49,7 @@ open class ERC20Balance {
         apis.request(1)
         return Flux.from(apis)
             .flatMap {
-                getBalance(it.cast(EthereumPosRpcUpstream::class.java), token, address)
+                getBalance(it.cast(EthereumLikeRpcUpstream::class.java), token, address)
             }
             .doOnNext {
                 apis.resolve()
@@ -57,7 +57,7 @@ open class ERC20Balance {
             .next()
     }
 
-    open fun getBalance(upstream: EthereumPosRpcUpstream, token: ERC20Token, address: Address): Mono<BigInteger> {
+    open fun getBalance(upstream: EthereumLikeRpcUpstream, token: ERC20Token, address: Address): Mono<BigInteger> {
         return upstream
             .getIngressReader()
             .read(prepareEthCall(token, address, upstream.getHead()))

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt

@@ -32,19 +32,20 @@ import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 
-open class EthereumRpcUpstream(
+open class EthereumLikeRpcUpstream(
     id: String,
     hash: Byte,
     val chain: Chain,
     options: UpstreamsConfig.Options,
     role: UpstreamsConfig.UpstreamRole,
     targets: CallMethods?,
-    private val node: QuorumForLabels.QuorumItem?,
+    node: QuorumForLabels.QuorumItem?,
     connectorFactory: ConnectorFactory,
-    chainConfig: ChainsConfig.ChainConfig
+    chainConfig: ChainsConfig.ChainConfig,
-) : EthereumUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled {
+    skipEnhance: Boolean
+) : EthereumLikeUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled {
     private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
-    private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, false)
+    private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, skipEnhance)
 
     private var validatorSubscription: Disposable? = null
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeUpstream.kt

@@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.upstream.Capability
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 
-abstract class EthereumUpstream(
+abstract class EthereumLikeUpstream(
     id: String,
     hash: Byte,
     options: UpstreamsConfig.Options,

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt

@@ -47,7 +47,7 @@ import reactor.core.scheduler.Scheduler
 @Suppress("UNCHECKED_CAST")
 open class EthereumMultistream(
     chain: Chain,
-    val upstreams: MutableList<EthereumUpstream>,
+    val upstreams: MutableList<EthereumLikeUpstream>,
     caches: Caches,
     private val headScheduler: Scheduler,
     tracer: Tracer,

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt

@@ -35,12 +35,14 @@ class EthereumRpcHead(
 ) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator, headScheduler), Lifecycle {
 
     private var refreshSubscription: Disposable? = null
+    private var isSyncing = false
 
     override fun start() {
         super.start()
         refreshSubscription?.dispose()
         val base = Flux.interval(interval)
             .publishOn(headScheduler)
+            .filter { !isSyncing }
             .flatMap {
                 getLatestBlock(api)
             }
@@ -51,6 +53,10 @@ class EthereumRpcHead(
         return refreshSubscription != null
     }
 
+    override fun onSyncingNode(isSyncing: Boolean) {
+        this.isSyncing = isSyncing
+    }
+
     override fun stop() {
         super.stop()
         refreshSubscription?.dispose()

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt

@@ -65,8 +65,7 @@ open class EthereumUpstreamValidator(
         if (!options.validateSyncing) {
             return Mono.just(UpstreamAvailability.OK)
         }
-        return upstream
+        return upstream.getIngressReader()
-            .getIngressReader()
             .read(JsonRpcRequest("eth_syncing", listOf()))
             .flatMap(JsonRpcResponse::requireResult)
             .map { objectMapper.readValue(it, SyncingJson::class.java) }
@@ -75,8 +74,10 @@ open class EthereumUpstreamValidator(
                 Mono.fromCallable { log.warn("No response for eth_syncing from ${upstream.getId()}") }
                     .then(Mono.error(TimeoutException("Validation timeout for Syncing")))
             )
-            .map { value ->
+            .map {
-                if (value.isSyncing) {
+                val isSyncing = it.isSyncing
+                upstream.getHead().onSyncingNode(isSyncing)
+                if (isSyncing) {
                     UpstreamAvailability.SYNCING
                 } else {
                     UpstreamAvailability.OK

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt

@@ -50,6 +50,7 @@ class EthereumWsHead(
     private var connectionId: String? = null
     private var subscribed = false
     private var connected = false
+    private var isSyncing = false
 
     private var subscription: Disposable? = null
     private val noHeadUpdatesSink = Sinks.many().multicast().directBestEffort<Boolean>()
@@ -78,6 +79,13 @@ class EthereumWsHead(
         noHeadUpdatesSink.tryEmitNext(true)
     }
 
+    override fun onSyncingNode(isSyncing: Boolean) {
+        if (isSyncing && !this.isSyncing) {
+            cancelSub()
+        }
+        this.isSyncing = isSyncing
+    }
+
     fun listenNewHeads(): Flux<BlockContainer> {
         return subscribe()
             .map {
@@ -131,8 +139,7 @@ class EthereumWsHead(
 
     override fun stop() {
         super.stop()
-        subscription?.dispose()
+        cancelSub()
-        subscription = null
         noHeadUpdatesSink.tryEmitComplete()
     }
 
@@ -168,10 +175,16 @@ class EthereumWsHead(
             noHeadUpdatesSink.asFlux(),
             connectionStates,
         ).subscribeOn(wsConnectionResubscribeScheduler)
-            .filter { it && !subscribed && connected }
+            .filter { it && !subscribed && connected && !isSyncing }
             .subscribe {
                 log.warn("Restart ws head, upstreamId: $upstreamId")
                 start()
             }
     }
+
+    private fun cancelSub() {
+        subscription?.dispose()
+        subscription = null
+        subscribed = false
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt

@@ -45,7 +45,7 @@ import reactor.core.scheduler.Scheduler
 @Suppress("UNCHECKED_CAST")
 open class EthereumPosMultiStream(
     chain: Chain,
-    val upstreams: MutableList<EthereumPosUpstream>,
+    val upstreams: MutableList<EthereumLikeUpstream>,
     caches: Caches,
     private val headScheduler: Scheduler,
     tracer: Tracer

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt

@@ -1,103 +0,0 @@
-/**
- * Copyright (c) 2020 EmeraldPay, Inc
- * Copyright (c) 2019 ETCDEV GmbH
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *        http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package io.emeraldpay.dshackle.upstream.ethereum
-
-import io.emeraldpay.dshackle.Chain
-import io.emeraldpay.dshackle.cache.Caches
-import io.emeraldpay.dshackle.cache.CachesEnabled
-import io.emeraldpay.dshackle.config.ChainsConfig
-import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.JsonRpcReader
-import io.emeraldpay.dshackle.startup.QuorumForLabels
-import io.emeraldpay.dshackle.upstream.Head
-import io.emeraldpay.dshackle.upstream.Lifecycle
-import io.emeraldpay.dshackle.upstream.Upstream
-import io.emeraldpay.dshackle.upstream.UpstreamAvailability
-import io.emeraldpay.dshackle.upstream.calls.CallMethods
-import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory
-import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
-import reactor.core.Disposable
-
-open class EthereumPosRpcUpstream(
-    id: String,
-    hash: Byte,
-    val chain: Chain,
-    options: UpstreamsConfig.Options,
-    role: UpstreamsConfig.UpstreamRole,
-    targets: CallMethods?,
-    node: QuorumForLabels.QuorumItem?,
-    connectorFactory: ConnectorFactory,
-    chainConfig: ChainsConfig.ChainConfig
-) : EthereumPosUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled {
-    private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
-    private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, true)
-
-    private var validatorSubscription: Disposable? = null
-
-    override fun setCaches(caches: Caches) {
-        if (connector is CachesEnabled) {
-            connector.setCaches(caches)
-        }
-    }
-
-    override fun start() {
-        log.info("Configured for ${chain.chainName}")
-        connector.start()
-        if (getOptions().disableValidation) {
-            log.warn("Disable validation for upstream ${this.getId()}")
-            this.setLag(0)
-            this.setStatus(UpstreamAvailability.OK)
-        } else {
-            log.debug("Start validation for upstream ${this.getId()}")
-            validatorSubscription = validator.start()
-                .subscribe(this::setStatus)
-        }
-    }
-    override fun getHead(): Head {
-        return connector.getHead()
-    }
-
-    override fun stop() {
-        validatorSubscription?.dispose()
-        validatorSubscription = null
-        connector.stop()
-    }
-
-    override fun isRunning(): Boolean {
-        return connector.isRunning()
-    }
-
-    override fun getIngressReader(): JsonRpcReader {
-        return connector.getIngressReader()
-    }
-
-    override fun isGrpc(): Boolean {
-        return false
-    }
-
-    @Suppress("UNCHECKED_CAST")
-    override fun <T : Upstream> cast(selfType: Class<T>): T {
-        if (!selfType.isAssignableFrom(this.javaClass)) {
-            throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
-        }
-        return this as T
-    }
-
-    override fun getIngressSubscription(): EthereumIngressSubscription {
-        return connector.getIngressSubscription()
-    }
-}

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt

@@ -1,51 +0,0 @@
-/**
- * Copyright (c) 2020 EmeraldPay, Inc
- * Copyright (c) 2019 ETCDEV GmbH
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *        http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package io.emeraldpay.dshackle.upstream.ethereum
-
-import io.emeraldpay.dshackle.config.ChainsConfig
-import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.startup.QuorumForLabels
-import io.emeraldpay.dshackle.upstream.Capability
-import io.emeraldpay.dshackle.upstream.DefaultUpstream
-import io.emeraldpay.dshackle.upstream.calls.CallMethods
-
-abstract class EthereumPosUpstream(
-    id: String,
-    hash: Byte,
-    options: UpstreamsConfig.Options,
-    role: UpstreamsConfig.UpstreamRole,
-    targets: CallMethods?,
-    private val node: QuorumForLabels.QuorumItem?,
-    chainConfig: ChainsConfig.ChainConfig
-) : DefaultUpstream(id, hash, options, role, targets, node, chainConfig) {
-
-    private val capabilities = if (options.providesBalance != false) {
-        setOf(Capability.RPC, Capability.BALANCE)
-    } else {
-        setOf(Capability.RPC)
-    }
-
-    override fun getCapabilities(): Set<Capability> {
-        return capabilities
-    }
-
-    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
-        return node?.let { listOf(it.labels) } ?: emptyList()
-    }
-
-    abstract fun getIngressSubscription(): EthereumIngressSubscription
-}

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt

@@ -34,7 +34,7 @@ import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
@@ -62,7 +62,7 @@ open class EthereumGrpcUpstream(
     overrideLabels: UpstreamsConfig.Labels?,
     chainConfig: ChainsConfig.ChainConfig,
     headScheduler: Scheduler,
-) : EthereumUpstream(
+) : EthereumLikeUpstream(
     "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
     hash,
     UpstreamsConfig.PartialOptions.getDefaults().buildOptions(),

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt

@@ -33,7 +33,7 @@ import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.Upstream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription
 import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
@@ -56,7 +56,7 @@ open class EthereumPosGrpcUpstream(
     overrideLabels: UpstreamsConfig.Labels?,
     chainConfig: ChainsConfig.ChainConfig,
     headScheduler: Scheduler,
-) : EthereumPosUpstream(
+) : EthereumLikeUpstream(
     "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
     hash,
     UpstreamsConfig.PartialOptions.getDefaults().buildOptions(),

src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy

@@ -26,9 +26,9 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock
 import io.emeraldpay.dshackle.test.MultistreamHolderMock
 import io.emeraldpay.dshackle.test.TestingCommons
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream
-import io.emeraldpay.etherjar.domain.BlockHash
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
+import io.emeraldpay.etherjar.domain.BlockHash
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import reactor.core.publisher.Mono
 import reactor.test.StepVerifier
@@ -43,7 +43,7 @@ class StreamHeadSpec extends Specification {
 
     def "Errors on unavailable chain"() {
         setup:
-        def upstreams = new MultistreamHolderMock(Chain.ETHEREUM__MAINNET, Stub(EthereumPosRpcUpstream))
+        def upstreams = new MultistreamHolderMock(Chain.ETHEREUM__MAINNET, Stub(EthereumLikeRpcUpstream))
         def streamHead = new StreamHead(upstreams)
         when:
         def flux = streamHead.add(

src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy

@@ -79,7 +79,7 @@ class EthereumHeadMock implements Head {
     }
 
     @Override
-    void onNoHeadUpdates() {
+    void onSyncingNode(boolean isSyncing) {
 
     }
 }

src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy

@@ -24,19 +24,14 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
-import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
+import io.emeraldpay.dshackle.upstream.calls.*
-import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream
-import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
-import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
-import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.jetbrains.annotations.NotNull
 import org.reactivestreams.Publisher
 
-
+class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream {
-class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
     EthereumHeadMock ethereumHeadMock
 
 
@@ -75,7 +70,9 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
                 methods,
                 new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)),
                 new ConnectorFactoryMock(api, new EthereumHeadMock()),
-                ChainsConfig.ChainConfig.default())
+                ChainsConfig.ChainConfig.default(),
+                true
+        )
         this.ethereumHeadMock = this.getHead() as EthereumHeadMock
         setLag(0)
         setStatus(UpstreamAvailability.OK)

src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy

@@ -23,18 +23,14 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
-import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods
+import io.emeraldpay.dshackle.upstream.calls.*
-import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream
-import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
-import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
-import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.jetbrains.annotations.NotNull
 import org.reactivestreams.Publisher
 
-class EthereumRpcUpstreamMock extends EthereumRpcUpstream {
+class EthereumRpcUpstreamMock extends EthereumLikeRpcUpstream {
     EthereumHeadMock ethereumHeadMock
 
 
@@ -65,7 +61,9 @@ class EthereumRpcUpstreamMock extends EthereumRpcUpstream {
                 methods,
                 new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()),
                 new ConnectorFactoryMock(api, new EthereumHeadMock()),
-                ChainsConfig.ChainConfig.default())
+                ChainsConfig.ChainConfig.default(),
+                false
+        )
         this.ethereumHeadMock = this.getHead() as EthereumHeadMock
         setLag(0)
         setStatus(UpstreamAvailability.OK)

src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy

@@ -25,10 +25,9 @@ import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream
 import org.jetbrains.annotations.NotNull
-import org.springframework.cloud.sleuth.Tracer
 import org.springframework.cloud.sleuth.brave.bridge.BraveTracer
 import reactor.core.scheduler.Schedulers
 
@@ -46,9 +45,9 @@ class MultistreamHolderMock implements MultistreamHolder {
             if (BlockchainType.from(chain) == BlockchainType.EVM_POS) {
                 if (up instanceof EthereumPosMultiStream) {
                     upstreams[chain] = up
-                } else if (up instanceof EthereumPosRpcUpstream) {
+                } else if (up instanceof EthereumLikeRpcUpstream) {
                     upstreams[chain] = new EthereumPosMultiStream(
-                            chain, [up as EthereumPosRpcUpstream], Caches.default(),
+                            chain, [up as EthereumLikeRpcUpstream], Caches.default(),
                             Schedulers.parallel(), TestingCommons.tracerMock()
                     )
                 } else {
@@ -97,15 +96,15 @@ class MultistreamHolderMock implements MultistreamHolder {
         CallMethods customMethods = null
         Head customHead = null
 
-        EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumPosRpcUpstream> upstreams, @NotNull Caches caches) {
+        EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumLikeRpcUpstream> upstreams, @NotNull Caches caches) {
             super(chain, upstreams, caches, Schedulers.parallel(), new BraveTracer(null, null, null))
         }
 
-        EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumPosRpcUpstream> upstreams) {
+        EthereumMultistreamMock(@NotNull Chain chain, @NotNull List<EthereumLikeRpcUpstream> upstreams) {
             this(chain, upstreams, Caches.default())
         }
 
-        EthereumMultistreamMock(@NotNull Chain chain, @NotNull EthereumPosRpcUpstream upstream) {
+        EthereumMultistreamMock(@NotNull Chain chain, @NotNull EthereumLikeRpcUpstream upstream) {
             this(chain, [upstream])
         }
 

src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy

@@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.test.EthereumApiStub
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import reactor.core.scheduler.Schedulers
@@ -42,7 +42,7 @@ class FilteredApisSpec extends Specification {
     def "Verifies labels"() {
         setup:
         def i = 0
-        List<EthereumRpcUpstream> upstreams = [
+        List<EthereumLikeRpcUpstream> upstreams = [
                 [test: "foo"],
                 [test: "bar"],
                 [test: "foo", test2: "baz"],
@@ -61,7 +61,7 @@ class FilteredApisSpec extends Specification {
                     Schedulers.parallel(),
                     Schedulers.parallel()
             )
-            new EthereumRpcUpstream(
+            new EthereumLikeRpcUpstream(
                     "test",
                     (byte) 123,
                     Chain.ETHEREUM__MAINNET,
@@ -70,7 +70,8 @@ class FilteredApisSpec extends Specification {
                     ethereumTargets,
                     new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)),
                     connectorFactory,
-                    ChainsConfig.ChainConfig.default()
+                    ChainsConfig.ChainConfig.default(),
+                    false
             )
         }
         def matcher = new Selector.LabelMatcher("test", ["foo"])

src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy

@@ -27,8 +27,8 @@ import io.emeraldpay.dshackle.startup.UpstreamChangeEvent
 import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
 import io.emeraldpay.dshackle.upstream.grpc.EthereumPosGrpcUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -252,7 +252,7 @@ class MultistreamSpec extends Specification {
         setup:
         def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"]))
         def up2 = new EthereumPosRpcUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
-        def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList<EthereumPosUpstream>(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock())
+        def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList<EthereumLikeRpcUpstream>(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock())
         when:
         ms.onUpstreamChange(
                 new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED)
@@ -305,7 +305,7 @@ class MultistreamSpec extends Specification {
         setup:
         def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"]))
         def up2 = new EthereumPosRpcUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"]))
-        def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList<EthereumPosUpstream>(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock())
+        def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList<EthereumLikeRpcUpstream>(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock())
         def head1 = createBlock(250, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448401b")
         def head2 = createBlock(270, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448402b")
         def head3 = createBlock(100, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448412b")
@@ -339,7 +339,7 @@ class MultistreamSpec extends Specification {
 
     class TestEthereumPosMultistream extends EthereumPosMultiStream {
 
-        TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<EthereumPosUpstream> upstreams, @NotNull Caches caches) {
+        TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List<EthereumLikeRpcUpstream> upstreams, @NotNull Caches caches) {
             super(chain, upstreams, caches, Schedulers.parallel(), TestingCommons.tracerMock())
         }
 

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy

@@ -47,7 +47,7 @@ class ERC20BalanceSpec extends Specification {
                         JsonRpcResponse.ok('"0x0000000000000000000000000000000000000000000000000000001f28d72868"')
                 )
 
-        EthereumPosRpcUpstream upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM__MAINNET, api)
+        EthereumLikeRpcUpstream upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM__MAINNET, api)
         ERC20Token token = new ERC20Token(Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9"))
         ERC20Balance query = new ERC20Balance()
 
@@ -73,7 +73,7 @@ class ERC20BalanceSpec extends Specification {
                         JsonRpcResponse.ok('"0x0000000000000000000000000000000000000000000000000000001f28d72868"')
                 )
 
-        EthereumPosRpcUpstream upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM__MAINNET, api)
+        EthereumLikeRpcUpstream upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM__MAINNET, api)
         ERC20Token token = new ERC20Token(Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9"))
         ERC20Balance query = new ERC20Balance()
 

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidatorSpec.groovy

@@ -16,9 +16,13 @@
 package io.emeraldpay.dshackle.upstream.ethereum
 
 import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.test.ApiReaderMock
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.etherjar.rpc.RpcResponseError
+import reactor.core.publisher.Mono
 import reactor.util.function.Tuples
 import spock.lang.Specification
 
@@ -30,7 +34,7 @@ class EthereumUpstreamValidatorSpec extends Specification {
 
     def "Resolve to final availability"() {
         setup:
-        def validator = new EthereumUpstreamValidator(Stub(EthereumUpstream), UpstreamsConfig.PartialOptions.getDefaults().buildOptions())
+        def validator = new EthereumUpstreamValidator(Stub(EthereumLikeUpstream), UpstreamsConfig.PartialOptions.getDefaults().buildOptions())
         expect:
         validator.resolve(Tuples.of(sync, peers)) == exp
         where:
@@ -51,7 +55,7 @@ class EthereumUpstreamValidatorSpec extends Specification {
         def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
             it.validateSyncing = false
         }.buildOptions()
-        def up = Mock(EthereumUpstream)
+        def up = Mock(EthereumLikeUpstream)
         def validator = new EthereumUpstreamValidator(up, options)
 
         when:
@@ -79,6 +83,33 @@ class EthereumUpstreamValidatorSpec extends Specification {
         act == OK
     }
 
+    def "Execute onSyncingNode with result of eth_syncing"() {
+        setup:
+        def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
+            it.validateSyncing = true
+        }.buildOptions()
+        def up = Mock(EthereumLikeUpstream) {
+            2 * getIngressReader() >> Mock(Reader) { reader ->
+                2 * reader.read(_) >>> [
+                        Mono.just(new JsonRpcResponse('true'.getBytes(), null)),
+                        Mono.just(new JsonRpcResponse('false'.getBytes(), null))
+                ]
+            }
+            2 * getHead() >> Mock(Head) {head ->
+                1 * head.onSyncingNode(true)
+                1 * head.onSyncingNode(false)
+            }
+        }
+        def validator = new EthereumUpstreamValidator(up, options)
+
+        when:
+        def act = validator.validateSyncing().block(Duration.ofSeconds(1))
+        def act2 = validator.validateSyncing().block(Duration.ofSeconds(1))
+        then:
+        act == SYNCING
+        act2 == OK
+    }
+
     def "Syncing is SYNCING when state returned from upstream"() {
         setup:
         def options = UpstreamsConfig.PartialOptions.getDefaults().tap {
@@ -121,7 +152,7 @@ class EthereumUpstreamValidatorSpec extends Specification {
             it.validatePeers = false
             it.minPeers = 10
         }.buildOptions()
-        def up = Mock(EthereumUpstream)
+        def up = Mock(EthereumLikeUpstream)
         def validator = new EthereumUpstreamValidator(up, options)
 
         when:
@@ -137,7 +168,7 @@ class EthereumUpstreamValidatorSpec extends Specification {
             it.validatePeers = true
             it.minPeers = 0
         }.buildOptions()
-        def up = Mock(EthereumUpstream)
+        def up = Mock(EthereumLikeUpstream)
         def validator = new EthereumUpstreamValidator(up, options)
 
         when:

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy

@@ -257,4 +257,69 @@ class EthereumWsHeadSpec extends Specification {
                 .thenCancel()
                 .verify(Duration.ofSeconds(1))
     }
+
+    def "Reset current subscription if upstream is syncing and then restore it"() {
+        setup:
+        def block = new BlockJson<TransactionRefJson>()
+        block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS)
+        block.number = 103
+        block.parentHash = parent
+        block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
+        def secondBlock = new BlockJson<TransactionRefJson>()
+        secondBlock.parentHash = parent
+        secondBlock.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS)
+        secondBlock.number = 105
+        secondBlock.hash = BlockHash.from("0x29229361dc5aa1ec66c323dc7a299e2b61a8c8dd2a3522d41255ec10eca25dd8")
+
+        def firstHeadBlock = block.with {
+            Global.objectMapper.writeValueAsBytes(it)
+        }
+        def secondHeadBlock = secondBlock.with {
+            Global.objectMapper.writeValueAsBytes(it)
+        }
+
+        def apiMock = TestingCommons.api()
+        def connectionInfoSink = Sinks.many().multicast().directBestEffort()
+        apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], null)
+        apiMock.answerOnce("eth_blockNumber", [], Mono.empty())
+        apiMock.answerOnce("eth_getBlockByHash", ["0x29229361dc5aa1ec66c323dc7a299e2b61a8c8dd2a3522d41255ec10eca25dd8", false], null)
+        apiMock.answerOnce("eth_blockNumber", [], Mono.empty())
+
+        def ws = Mock(WsSubscriptions) {
+            1 * it.connectionInfoFlux() >> connectionInfoSink.asFlux()
+            2 * subscribe("newHeads") >>> [
+                    new WsSubscriptions.SubscribeData(Flux.fromIterable([firstHeadBlock]), "id"),
+                    new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id"),
+            ]
+        }
+
+        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel())
+
+        when:
+        def act = head.getFlux()
+
+        then:
+        StepVerifier.create(act)
+                .then { head.start() }
+                .expectNext(BlockContainer.from(block))
+                .then {
+                    head.onSyncingNode(true)
+                }
+                .then {
+                    assert !head.isRunning()
+                }
+                .then {
+                    head.onNoHeadUpdates()
+                }
+                .then {
+                    assert !head.isRunning()
+                }
+                .then {
+                    head.onSyncingNode(false)
+                    head.onNoHeadUpdates()
+                }
+                .expectNext(BlockContainer.from(secondBlock))
+                .thenCancel()
+                .verify(Duration.ofSeconds(1))
+    }
 }