diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt index 52404ccb..00f57115 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/data/BlockContainer.kt +++ b/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, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 67424830..0c6baa8b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/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) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index f3182f2a..ea191a3c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt index f3262450..9f368e41 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt @@ -84,6 +84,7 @@ open class EthereumUpstreamValidator( } .onErrorReturn(UpstreamAvailability.UNAVAILABLE) } + fun validatePeers(): Mono { if (!options.validatePeers || options.minPeers == 0) { return Mono.just(UpstreamAvailability.OK) @@ -110,8 +111,10 @@ open class EthereumUpstreamValidator( } fun start(): Flux { - return Flux.interval(Duration.ofSeconds(options.validationInterval.toLong())) - .subscribeOn(scheduler) + return Flux.interval( + Duration.ZERO, + Duration.ofSeconds(options.validationInterval.toLong()), + ).subscribeOn(scheduler) .flatMap { validate() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index 9bbe36b6..8666ee90 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/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 || - block.transactions == null || - block.transactions.isEmpty() || - block.totalDifficulty == null + if (!skipEnhance && ( + block.difficulty == null || + block.transactions == null || + block.transactions.isEmpty() || + block.totalDifficulty == null + ) ) { enhanceRealBlock(block) } else { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt index 5dc4100e..15ccbd52 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/ConnectorFactory.kt +++ b/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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt index c307bde5..9cf9b8f9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt +++ b/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 + ) } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt index c29d5b50..cfdb0bb0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt +++ b/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 diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt index c965cc79..d8e87c07 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt +++ b/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) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt index 89e2d2d7..03fc0b45 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/EthereumWsIngressSubscription.kt +++ b/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 { @@ -44,7 +39,7 @@ class EthereumWsIngressSubscription( return null } - override fun getPendingTxes(): PendingTxesSource? { + override fun getPendingTxes(): PendingTxesSource { return pendingTxes } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt index 51f0b86c..3808d238 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt +++ b/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 diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy index abc947d3..812194d7 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/ConnectorFactoryMock.groovy +++ b/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) } } \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy index 510167c4..3c0ed26d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy +++ b/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 api, CallMethods methods, Map 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) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy index 653788be..9bf4286f 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy +++ b/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()