add grpc support for PoS Ethereum

2022-07-29 17:42:19 +04:00
parent eb95c591f7
commit 1df7221c68
24 changed files with 302 additions and 230 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
@@ -102,6 +102,7 @@ open class UpstreamsConfig {
        var host: String? = null
        var port: Int = 0
        var auth: AuthConfig.ClientTlsAuth? = null
+        var nodeRating: Int = 0
    }

    class EthereumConnection : RpcConnection() {
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
@@ -106,6 +106,9 @@ class UpstreamsConfigReader(
                    config.upstreams.add(upstream)
                    val connection = UpstreamsConfig.GrpcConnection()
                    upstream.connection = connection
+                    getValueAsInt(connConfigNode, "node-rating")?.let {
+                        connection.nodeRating = it
+                    }
                    getValueAsString(connConfigNode, "host")?.let {
                        connection.host = it
                    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
@@ -29,8 +29,8 @@ import io.emeraldpay.dshackle.upstream.bitcoin.ExtractBlock
 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.EthereumPosUpstream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
@@ -160,7 +160,7 @@ open class ConfiguredUpstreams(
        if (connectorFactory == null) {
            return null
        }
-        val upstream = EthereumPosUpstream(
+        val upstream = EthereumPosRpcUpstream(
            config.id!!,
            chain,
            options, config.role,
@@ -218,7 +218,7 @@ open class ConfiguredUpstreams(
        config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
        chain: Chain,
        options: UpstreamsConfig.Options
-    ) : EthereumUpstream? {
+    ) : EthereumRpcUpstream? {
        val conn = config.connection!!

        val urls = ArrayList<URI>()
@@ -228,7 +228,7 @@ open class ConfiguredUpstreams(
        if (connectorFactory == null) {
            return null
        }
-        val upstream = EthereumUpstream(
+        val upstream = EthereumRpcUpstream(
            config.id!!,
            chain,
            options, config.role,
@@ -251,7 +251,8 @@ open class ConfiguredUpstreams(
            endpoint.host!!,
            endpoint.port,
            endpoint.auth,
-            fileResolver
+            fileResolver,
+            endpoint.nodeRating
        ).apply {
            timeout = options.timeout
        }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
@@ -24,10 +24,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
 import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultistream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.*
 import io.emeraldpay.grpc.BlockchainType
 import io.emeraldpay.grpc.Chain
 import org.slf4j.LoggerFactory
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt
+++ b/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(EthereumUpstream::class.java), token, address)
+                getBalance(it.cast(EthereumRpcUpstream::class.java), token, address)
            }
            .doOnNext {
                apis.resolve()
@@ -57,7 +57,7 @@ open class ERC20Balance {
            .next()
    }

-    open fun getBalance(upstream: EthereumUpstream, token: ERC20Token, address: Address): Mono<BigInteger> {
+    open fun getBalance(upstream: EthereumRpcUpstream, token: ERC20Token, address: Address): Mono<BigInteger> {
        return upstream
            .getApi()
            .read(prepareEthCall(token, address, upstream.getHead()))
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt
@@ -0,0 +1,98 @@
+/**
+ * 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.cache.Caches
+import io.emeraldpay.dshackle.cache.CachesEnabled
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.*
+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.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.grpc.Chain
+import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+import reactor.core.Disposable
+
+open class EthereumRpcUpstream(
+    id: String,
+    val chain: Chain,
+    options: UpstreamsConfig.Options,
+    role: UpstreamsConfig.UpstreamRole,
+    targets: CallMethods?,
+    private val node: QuorumForLabels.QuorumItem?,
+    connectorFactory: ConnectorFactory
+) : EthereumUpstream(id, options, role, targets, node), 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 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 != null && 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 getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+        return connector.getApi()
+    }
+
+    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
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
@@ -16,78 +16,19 @@
 */
 package io.emeraldpay.dshackle.upstream.ethereum

-import io.emeraldpay.dshackle.cache.Caches
-import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
-import io.emeraldpay.dshackle.upstream.*
+import io.emeraldpay.dshackle.upstream.Capability
+import io.emeraldpay.dshackle.upstream.DefaultUpstream
 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.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
-import io.emeraldpay.grpc.Chain
-import org.slf4j.LoggerFactory
-import org.springframework.context.Lifecycle
-import reactor.core.Disposable

-open class EthereumUpstream(
+abstract class EthereumUpstream(
    id: String,
-    val chain: Chain,
    options: UpstreamsConfig.Options,
    role: UpstreamsConfig.UpstreamRole,
    targets: CallMethods?,
-    private val node: QuorumForLabels.QuorumItem?,
-    connectorFactory: ConnectorFactory
-) : DefaultUpstream(id, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
-    private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)
-    private val validator : EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
-    private val connector : EthereumConnector = connectorFactory.create(this, validator, chain)
-
-    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 != null && 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 getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
-        return connector.getApi()
-    }
-
-    override fun isGrpc(): Boolean {
-        return false
-    }
+    private val node: QuorumForLabels.QuorumItem?
+) : DefaultUpstream(id, options, role, targets, node) {

    private val capabilities = if (options.providesBalance != false) {
        setOf(Capability.RPC, Capability.BALANCE)
@@ -102,12 +43,4 @@ open class EthereumUpstream(
    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
        return node?.let { listOf(it.labels) } ?: emptyList()
    }
-
-    @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
-    }
-}
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt
@@ -25,7 +25,6 @@ import io.emeraldpay.dshackle.upstream.MergedHead
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.Selector
 import io.emeraldpay.dshackle.upstream.Upstream
-import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
 import io.emeraldpay.dshackle.upstream.forkchoice.PriorityForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
@@ -0,0 +1,98 @@
+/**
+ * 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.cache.Caches
+import io.emeraldpay.dshackle.cache.CachesEnabled
+import io.emeraldpay.dshackle.config.UpstreamsConfig
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.startup.QuorumForLabels
+import io.emeraldpay.dshackle.upstream.*
+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.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.grpc.Chain
+import org.slf4j.LoggerFactory
+import org.springframework.context.Lifecycle
+import reactor.core.Disposable
+
+open class EthereumPosRpcUpstream(
+    id: String,
+    val chain: Chain,
+    options: UpstreamsConfig.Options,
+    role: UpstreamsConfig.UpstreamRole,
+    targets: CallMethods?,
+    private val node: QuorumForLabels.QuorumItem?,
+    connectorFactory: ConnectorFactory
+) : EthereumPosUpstream(id, options, role, targets, node), 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 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 != null && 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 getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
+        return connector.getApi()
+    }
+
+    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
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt
@@ -16,78 +16,19 @@
 */
 package io.emeraldpay.dshackle.upstream.ethereum

-import io.emeraldpay.dshackle.cache.Caches
-import io.emeraldpay.dshackle.cache.CachesEnabled
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
-import io.emeraldpay.dshackle.upstream.*
+import io.emeraldpay.dshackle.upstream.Capability
+import io.emeraldpay.dshackle.upstream.DefaultUpstream
 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.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
-import io.emeraldpay.grpc.Chain
-import org.slf4j.LoggerFactory
-import org.springframework.context.Lifecycle
-import reactor.core.Disposable

-open class EthereumPosUpstream(
+abstract class EthereumPosUpstream(
    id: String,
-    val chain: Chain,
    options: UpstreamsConfig.Options,
    role: UpstreamsConfig.UpstreamRole,
    targets: CallMethods?,
-    private val node: QuorumForLabels.QuorumItem?,
-    connectorFactory: ConnectorFactory
-) : DefaultUpstream(id, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
-    private val log = LoggerFactory.getLogger(EthereumPosUpstream::class.java)
-    private val validator : EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
-    private val connector : EthereumConnector = connectorFactory.create(this, validator, chain)
-
-    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 != null && 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 getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
-        return connector.getApi()
-    }
-
-    override fun isGrpc(): Boolean {
-        return false
-    }
+    private val node: QuorumForLabels.QuorumItem?
+) : DefaultUpstream(id, options, role, targets, node) {

    private val capabilities = if (options.providesBalance != false) {
        setOf(Capability.RPC, Capability.BALANCE)
@@ -102,12 +43,4 @@ open class EthereumPosUpstream(
    override fun getLabels(): Collection<UpstreamsConfig.Labels> {
        return node?.let { listOf(it.labels) } ?: emptyList()
    }
-
-    @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
-    }
-}
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
@@ -50,7 +50,7 @@ open class EthereumGrpcUpstream(
    private val chain: Chain,
    private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
    private val client: JsonRpcGrpcClient
-) : DefaultUpstream(
+) : EthereumUpstream(
    "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
    UpstreamsConfig.Options.getDefaults(),
    role,
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
@@ -26,8 +26,8 @@ import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.startup.QuorumForLabels
 import io.emeraldpay.dshackle.upstream.*
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
-import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
-import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
+import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
@@ -48,9 +48,10 @@ open class EthereumPosGrpcUpstream(
    private val parentId: String,
    role: UpstreamsConfig.UpstreamRole,
    private val chain: Chain,
-    private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
-    private val client: JsonRpcGrpcClient
-) : DefaultUpstream(
+    remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
+    client: JsonRpcGrpcClient,
+    nodeRating: Int
+) : EthereumPosUpstream(
    "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
    UpstreamsConfig.Options.getDefaults(),
    role,
@@ -94,7 +95,7 @@ open class EthereumPosGrpcUpstream(

    private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java)
    private val upstreamStatus = GrpcUpstreamStatus()
-    private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
+    private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, NoChoiceWithPriorityForkChoice(nodeRating))
    private var capabilities: Set<Capability> = emptySet()

    private val defaultReader: Reader<JsonRpcRequest, JsonRpcResponse> = client.forSelector(Selector.empty)
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
@@ -56,7 +56,8 @@ class GrpcUpstreams(
    private val host: String,
    private val port: Int,
    private val auth: AuthConfig.ClientTlsAuth? = null,
-    private val fileResolver: FileResolver
+    private val fileResolver: FileResolver,
+    private val nodeRating: Int
 ) {
    private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)

@@ -192,6 +193,8 @@ class GrpcUpstreams(
            return getOrCreateEthereum(chain, metrics)
        } else if (blockchainType == BlockchainType.BITCOIN) {
            return getOrCreateBitcoin(chain, metrics)
+        } else if (blockchainType == BlockchainType.ETHEREUM_POS) {
+            return getOrCreateEthereumPos(chain, metrics)
        } else {
            throw IllegalArgumentException("Unsupported blockchain: $chain")
        }
@@ -213,6 +216,22 @@ class GrpcUpstreams(
        }
    }

+    fun getOrCreateEthereumPos(chain: Chain, metrics: RpcMetrics): UpstreamChange {
+        lock.withLock {
+            val current = known[chain]
+            return if (current == null) {
+                val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
+                val created = EthereumPosGrpcUpstream(id, role, chain, client!!, rpcClient, nodeRating)
+                created.timeout = this.timeout
+                known[chain] = created
+                created.start()
+                UpstreamChange(chain, created, UpstreamChange.ChangeType.ADDED)
+            } else {
+                UpstreamChange(chain, current, UpstreamChange.ChangeType.REVALIDATED)
+            }
+        }
+    }
+
    fun getOrCreateBitcoin(chain: Chain, metrics: RpcMetrics): UpstreamChange {
        lock.withLock {
            val current = known[chain]