Upstream validation and enhancement improvement: (#146)

- now validation of upstream starts immediately after upstream starts - it allows it became available asap
- now we don't enrich PoS blocks which improves performance and decrease number of requests to node

src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt

@@ -42,7 +42,7 @@ class BlockContainer(
             return BlockContainer(
                 height = block.number,
                 hash = BlockId.from(block),
-                difficulty = block.totalDifficulty,
+                difficulty = block.totalDifficulty ?: BigInteger.ZERO,
                 timestamp = block.timestamp,
                 full = hasTransactions,
                 json = raw,

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

@@ -149,7 +149,6 @@ abstract class AbstractHead @JvmOverloads constructor(
 
     override fun stop() {
         stopping = true
-        log.debug("Stop ${this.javaClass.simpleName} $upstreamId")
         future?.let {
             it.cancel(true)
         }

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

@@ -46,7 +46,7 @@ open class EthereumRpcUpstream(
 ) : EthereumUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled {
     private val log = LoggerFactory.getLogger(EthereumRpcUpstream::class.java)
     private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
-    private val connector: EthereumConnector = connectorFactory.create(this, validator, chain)
+    private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, false)
 
     private var validatorSubscription: Disposable? = null
 

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

@@ -84,6 +84,7 @@ open class EthereumUpstreamValidator(
             }
             .onErrorReturn(UpstreamAvailability.UNAVAILABLE)
     }
+
     fun validatePeers(): Mono<UpstreamAvailability> {
         if (!options.validatePeers || options.minPeers == 0) {
             return Mono.just(UpstreamAvailability.OK)
@@ -110,8 +111,10 @@ open class EthereumUpstreamValidator(
     }
 
     fun start(): Flux<UpstreamAvailability> {
-        return Flux.interval(Duration.ofSeconds(options.validationInterval.toLong()))
+        return Flux.interval(
-            .subscribeOn(scheduler)
+            Duration.ZERO,
+            Duration.ofSeconds(options.validationInterval.toLong()),
+        ).subscribeOn(scheduler)
             .flatMap {
                 validate()
             }

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

@@ -28,7 +28,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.etherjar.rpc.json.BlockJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
-import org.slf4j.LoggerFactory
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
@@ -42,10 +41,9 @@ class EthereumWsHead(
     blockValidator: BlockValidator,
     private val api: JsonRpcReader,
     private val wsSubscriptions: WsSubscriptions,
+    private val skipEnhance: Boolean
 ) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator), Lifecycle {
 
-    private val log = LoggerFactory.getLogger(EthereumWsHead::class.java)
-
     private var subscription: Disposable? = null
 
     override fun isRunning(): Boolean {
@@ -71,10 +69,12 @@ class EthereumWsHead(
             .flatMap { block ->
                 // newHeads returns incomplete blocks, i.e. without some fields and without transaction hashes,
                 // so we need to fetch the full block data
-                if (block.difficulty == null ||
+                if (!skipEnhance && (
-                    block.transactions == null ||
+                    block.difficulty == null ||
-                    block.transactions.isEmpty() ||
+                        block.transactions == null ||
-                    block.totalDifficulty == null
+                        block.transactions.isEmpty() ||
+                        block.totalDifficulty == null
+                    )
                 ) {
                     enhanceRealBlock(block)
                 } else {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt

@@ -5,6 +5,12 @@ import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
 
 interface ConnectorFactory {
-    fun create(upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain): EthereumConnector
+    fun create(
+        upstream: DefaultUpstream,
+        validator: EthereumUpstreamValidator,
+        chain: Chain,
+        skipEnhance: Boolean
+    ): EthereumConnector
+
     fun isValid(): Boolean
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt

@@ -7,18 +7,14 @@ import io.emeraldpay.dshackle.upstream.HttpFactory
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstreamValidator
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
-import org.slf4j.LoggerFactory
 
 open class EthereumConnectorFactory(
     private val preferHttp: Boolean,
     private val wsFactory: EthereumWsConnectionPoolFactory?,
     private val httpFactory: HttpFactory?,
     private val forkChoice: ForkChoice,
-    private val blockValidator: BlockValidator
+    private val blockValidator: BlockValidator,
 ) : ConnectorFactory {
-    companion object {
-        private val log = LoggerFactory.getLogger(EthereumConnectorFactory::class.java)
-    }
 
     override fun isValid(): Boolean {
         if (preferHttp && httpFactory == null) {
@@ -27,13 +23,25 @@ open class EthereumConnectorFactory(
         return true
     }
 
-    override fun create(upstream: DefaultUpstream, validator: EthereumUpstreamValidator, chain: Chain): EthereumConnector {
+    override fun create(
+        upstream: DefaultUpstream,
+        validator: EthereumUpstreamValidator,
+        chain: Chain,
+        skipEnhance: Boolean
+    ): EthereumConnector {
         if (wsFactory != null && !preferHttp) {
-            return EthereumWsConnector(wsFactory, upstream, forkChoice, blockValidator)
+            return EthereumWsConnector(wsFactory, upstream, forkChoice, blockValidator, skipEnhance)
         }
         if (httpFactory == null) {
             throw java.lang.IllegalArgumentException("Can't create rpc connector if no http factory set")
         }
-        return EthereumRpcConnector(httpFactory.create(upstream.getId(), chain), wsFactory, upstream.getId(), forkChoice, blockValidator)
+        return EthereumRpcConnector(
+            httpFactory.create(upstream.getId(), chain),
+            wsFactory,
+            upstream.getId(),
+            forkChoice,
+            blockValidator,
+            skipEnhance
+        )
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt

@@ -24,7 +24,8 @@ class EthereumRpcConnector(
     wsFactory: EthereumWsConnectionPoolFactory?,
     id: String,
     forkChoice: ForkChoice,
-    blockValidator: BlockValidator
+    blockValidator: BlockValidator,
+    skipEnhance: Boolean
 ) : EthereumConnector, CachesEnabled {
     private val pool: WsConnectionPool?
     private val head: Head
@@ -38,9 +39,11 @@ class EthereumRpcConnector(
             // do not set upstream to the WS, since it doesn't control the RPC upstream
             pool = wsFactory.create(null)
             val subscriptions = WsSubscriptionsImpl(pool)
-            val wsHead = EthereumWsHead(id, AlwaysForkChoice(), blockValidator, getIngressReader(), subscriptions)
+            val wsHead =
+                EthereumWsHead(id, AlwaysForkChoice(), blockValidator, getIngressReader(), subscriptions, skipEnhance)
             // receive all new blocks through WebSockets, but also periodically verify with RPC in case if WS failed
-            val rpcHead = EthereumRpcHead(getIngressReader(), AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30))
+            val rpcHead =
+                EthereumRpcHead(getIngressReader(), AlwaysForkChoice(), id, blockValidator, Duration.ofSeconds(30))
             head = MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id")
         } else {
             pool = null

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt

@@ -17,7 +17,8 @@ class EthereumWsConnector(
     wsFactory: EthereumWsConnectionPoolFactory,
     upstream: DefaultUpstream,
     forkChoice: ForkChoice,
-    blockValidator: BlockValidator
+    blockValidator: BlockValidator,
+    skipEnhance: Boolean
 ) : EthereumConnector {
     private val pool: WsConnectionPool
     private val reader: JsonRpcReader
@@ -28,7 +29,7 @@ class EthereumWsConnector(
         pool = wsFactory.create(upstream)
         reader = JsonRpcWsClient(pool)
         val wsSubscriptions = WsSubscriptionsImpl(pool)
-        head = EthereumWsHead(upstream.getId(), forkChoice, blockValidator, reader, wsSubscriptions)
+        head = EthereumWsHead(upstream.getId(), forkChoice, blockValidator, reader, wsSubscriptions, skipEnhance)
         subscriptions = EthereumWsIngressSubscription(wsSubscriptions)
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt

@@ -20,16 +20,11 @@ import io.emeraldpay.dshackle.upstream.SubscriptionConnect
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumEgressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
 import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptions
-import org.slf4j.LoggerFactory
 
 class EthereumWsIngressSubscription(
-    private val conn: WsSubscriptions
+    conn: WsSubscriptions
 ) : IngressSubscription, EthereumIngressSubscription {
 
-    companion object {
-        private val log = LoggerFactory.getLogger(EthereumWsIngressSubscription::class.java)
-    }
-
     private val pendingTxes = WebsocketPendingTxes(conn)
 
     override fun getAvailableTopics(): List<String> {
@@ -44,7 +39,7 @@ class EthereumWsIngressSubscription(
         return null
     }
 
-    override fun getPendingTxes(): PendingTxesSource? {
+    override fun getPendingTxes(): PendingTxesSource {
         return pendingTxes
     }
 }

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

@@ -46,7 +46,7 @@ open class EthereumPosRpcUpstream(
 ) : EthereumPosUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled {
     private val log = LoggerFactory.getLogger(EthereumPosRpcUpstream::class.java)
     private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
-    private val connector: EthereumConnector = connectorFactory.create(this, validator, chain)
+    private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, false)
 
     private var validatorSubscription: Disposable? = null
 

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

@@ -23,7 +23,7 @@ class ConnectorFactoryMock implements ConnectorFactory {
         return true
     }
 
-    EthereumConnector create(DefaultUpstream upstream, EthereumUpstreamValidator validator, Chain chain)  {
+    EthereumConnector create(DefaultUpstream upstream, EthereumUpstreamValidator validator, Chain chain, boolean skipEnhance)  {
         return new EthereumConnectorMock(api, head)
     }
 }

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

@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.test
 import io.emeraldpay.dshackle.Chain
 import io.emeraldpay.dshackle.config.ChainsConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.config.UpstreamsConfig.Options
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
@@ -34,6 +35,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import org.jetbrains.annotations.NotNull
 import org.reactivestreams.Publisher
 
+
 class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
     EthereumHeadMock ethereumHeadMock
 
@@ -68,7 +70,7 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
 
     EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods, Map<String, String> labels) {
         super(id, (byte)id.hashCode(), chain,
-                UpstreamsConfig.Options.getDefaults(),
+                getOpts(),
                 UpstreamsConfig.UpstreamRole.PRIMARY,
                 methods,
                 new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)),
@@ -80,6 +82,12 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
         start()
     }
 
+    static Options getOpts() {
+        def opt = UpstreamsConfig.Options.getDefaults()
+        opt.setDisableValidation(true)
+        return opt
+    }
+
     void nextBlock(BlockContainer block) {
         this.ethereumHeadMock.nextBlock(block)
     }

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

@@ -56,7 +56,7 @@ class EthereumWsHeadSpec extends Specification {
 
         def ws = Mock(WsSubscriptions)
 
-        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws)
+        def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, false)
 
         when:
         def act = head.listenNewHeads().blockFirst()