Auto detect labels (#245)

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

@@ -23,6 +23,7 @@ import java.net.URI
 import java.time.Duration
 import java.util.Arrays
 import java.util.Locale
+import java.util.concurrent.ConcurrentHashMap
 
 open class UpstreamsConfig {
     var defaultOptions: MutableList<DefaultOptions> = ArrayList<DefaultOptions>()
@@ -188,7 +189,7 @@ open class UpstreamsConfig {
     }
 
     // TODO make it unmodifiable after initial load
-    class Labels : HashMap<String, String>() {
+    class Labels : ConcurrentHashMap<String, String>() {
 
         companion object {
             @JvmStatic

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

@@ -34,7 +34,7 @@ abstract class DefaultUpstream(
     private val options: UpstreamsConfig.Options,
     private val role: UpstreamsConfig.UpstreamRole,
     private val targets: CallMethods?,
-    node: QuorumForLabels.QuorumItem?,
+    private val node: QuorumForLabels.QuorumItem?,
     private val chainConfig: ChainsConfig.ChainConfig
 ) : Upstream {
 
@@ -161,11 +161,9 @@ abstract class DefaultUpstream(
 
     override fun nodeId(): Byte = hash
 
-    private val quorumByLabel = node?.let { QuorumForLabels(it) }
-        ?: QuorumForLabels(QuorumForLabels.QuorumItem.empty())
-
     open fun getQuorumByLabel(): QuorumForLabels {
-        return quorumByLabel
+        return node?.let { QuorumForLabels(it) }
+            ?: QuorumForLabels(QuorumForLabels.QuorumItem.empty())
     }
 
     override fun getId(): String {

src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt

@@ -92,7 +92,8 @@ class DefaultEthereumMethods(
         "eth_getStorageAt",
         "eth_getCode",
         "eth_getUncleByBlockHashAndIndex",
-        "eth_getLogs"
+        "eth_getLogs",
+        "eth_maxPriorityFeePerGas"
     )
 
     private val specialMethods = listOf(
@@ -224,10 +225,12 @@ class DefaultEthereumMethods(
     }
 
     private fun chainUnsupportedMethods(chain: Chain): Set<String> {
-        if (chain == Chain.OPTIMISM__MAINNET) {
+        return when (chain) {
-            return setOf("eth_getAccounts")
+            Chain.OPTIMISM__MAINNET -> setOf("eth_getAccounts")
+            Chain.ZKSYNC__MAINNET, Chain.ZKSYNC__TESTNET, Chain.POLYGON_ZKEVM__TESTNET, Chain.POLYGON_ZKEVM__MAINNET ->
+                setOf("eth_maxPriorityFeePerGas")
+            else -> emptySet()
         }
-        return emptySet()
     }
 
     override fun isCallable(method: String): Boolean {

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

@@ -29,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumLabelsDetector
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
 
@@ -39,13 +40,14 @@ open class EthereumLikeRpcUpstream(
     options: UpstreamsConfig.Options,
     role: UpstreamsConfig.UpstreamRole,
     targets: CallMethods?,
-    node: QuorumForLabels.QuorumItem?,
+    private val node: QuorumForLabels.QuorumItem?,
     connectorFactory: ConnectorFactory,
     chainConfig: ChainsConfig.ChainConfig,
     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, skipEnhance)
+    private val labelsDetector = EthereumLabelsDetector(this.getIngressReader())
 
     private var validatorSubscription: Disposable? = null
 
@@ -67,6 +69,13 @@ open class EthereumLikeRpcUpstream(
             validatorSubscription = validator.start()
                 .subscribe(this::setStatus)
         }
+        labelsDetector.detectLabels()
+            .toStream()
+            .forEach {
+                node?.labels?.let { labels ->
+                    labels[it.first] = it.second
+                }
+            }
     }
 
     override fun getIngressSubscription(): EthereumIngressSubscription {

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

@@ -0,0 +1,61 @@
+package io.emeraldpay.dshackle.upstream.ethereum.subscribe
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.module.kotlin.readValue
+import io.emeraldpay.dshackle.Global.Companion.objectMapper
+import io.emeraldpay.dshackle.reader.JsonRpcReader
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+class EthereumLabelsDetector(
+    private val reader: JsonRpcReader
+) {
+
+    fun detectLabels(): Flux<Pair<String, String>> {
+        return Flux.merge(
+            detectNodeType(),
+            detectArchiveNode()
+        )
+    }
+
+    private fun detectNodeType(): Mono<Pair<String, String>?> {
+        return reader
+            .read(JsonRpcRequest("web3_clientVersion", listOf()))
+            .flatMap(JsonRpcResponse::requireResult)
+            .mapNotNull {
+                val node = objectMapper.readValue<JsonNode>(it)
+                if (node.isTextual) {
+                    nodeType(node.textValue())?.run {
+                        "client_type" to this
+                    }
+                } else {
+                    null
+                }
+            }
+            .onErrorResume { Mono.empty() }
+    }
+
+    private fun detectArchiveNode(): Mono<Pair<String, String>> {
+        return reader
+            .read(JsonRpcRequest("eth_getBalance", listOf("0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", "0x1")))
+            .flatMap(JsonRpcResponse::requireResult)
+            .map { "archive" to "true" }
+            .onErrorResume { Mono.empty() }
+    }
+
+    private fun nodeType(nodeType: String): String? {
+        return if (nodeType.contains("erigon", true)) {
+            "erigon"
+        } else if (nodeType.contains("geth", true)) {
+            "geth"
+        } else if (nodeType.contains("bor", true)) {
+            "bor"
+        } else if (nodeType.contains("nethermind", true)) {
+            "nethermind"
+        } else {
+            null
+        }
+    }
+}

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

@@ -0,0 +1,65 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.test.ApiReaderMock
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.DefaultUpstream
+import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumLabelsDetector
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import kotlin.Pair
+import reactor.core.publisher.Mono
+import reactor.test.StepVerifier
+import spock.lang.Specification
+
+import java.time.Duration
+
+class EthereumLabelsDetectorSpec extends Specification {
+
+    def "Detect labels"() {
+        setup:
+        def up = TestingCommons.upstream(
+                new ApiReaderMock().tap {
+                    answer("web3_clientVersion", [], response)
+                    answer("eth_getBalance", ["0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", "0x1"], "")
+                }
+        )
+        def detector = new EthereumLabelsDetector(up.getIngressReader())
+
+        when:
+        def act = detector.detectLabels()
+        then:
+        StepVerifier.create(act)
+            .expectNext(
+                    new Pair<String, String>("client_type", clientType),
+                    new Pair<String, String>("archive", "true")
+            )
+            .expectComplete()
+            .verify(Duration.ofSeconds(1))
+        where:
+        response                                                | clientType
+        "Nethermind/v1.19.3+e8ac1da4/linux-x64/dotnet7.0.8"     | "nethermind"
+        "Geth/v1.12.0-stable-e501b3b0/linux-amd64/go1.20.3"     | "geth"
+        "Erigon/v1.12.0-stable-e501b3b0/linux-amd64/go1.20.3"   | "erigon"
+        "Bor/v0.4.0/linux-amd64/go1.19.10"                      | "bor"
+    }
+
+    def "No any label"() {
+        setup:
+        def up = Mock(DefaultUpstream) {
+            1 * getIngressReader() >> Mock(Reader) {
+                1 * read(new JsonRpcRequest("web3_clientVersion", [])) >>
+                        Mono.just(new JsonRpcResponse('no/v1.19.3+e8ac1da4/linux-x64/dotnet7.0.8'.getBytes(), null))
+                1 * read(new JsonRpcRequest("eth_getBalance", ["0x756F45E3FA69347A9A973A725E3C98bC4db0b5a0", "0x1"])) >>
+                        Mono.error(new RuntimeException())
+            }
+        }
+        def detector = new EthereumLabelsDetector(up.getIngressReader())
+        when:
+        def act = detector.detectLabels()
+        then:
+        StepVerifier.create(act)
+            .expectComplete()
+            .verify(Duration.ofSeconds(1))
+    }
+}