From 00a1490726a24171c027f91aa20304587a9ecfa2 Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Mon, 27 Feb 2023 14:22:50 +0500 Subject: [PATCH] Upstream configuration improvement: (#149) - now we can specify connector mode for better fine tuning of connection --- docs/04-upstream-config.adoc | 21 +++++++ .../dshackle/config/UpstreamsConfig.kt | 18 ++++++ .../dshackle/config/UpstreamsConfigReader.kt | 3 + .../dshackle/startup/ConfiguredUpstreams.kt | 26 ++++---- .../upstream/ethereum/WsConnectionImpl.kt | 2 +- .../connectors/EthereumConnectorFactory.kt | 42 ++++++++++++- .../connectors/EthereumRpcConnector.kt | 37 ++++++++--- .../config/UpstreamsConfigReaderSpec.groovy | 28 ++++++++- .../dshackle/upstream/FilteredApisSpec.groovy | 2 +- .../dshackle/config/UpstreamsConfigTest.kt | 61 +++++++++++++++++++ .../configs/upstreams-connector-mode.yaml | 12 ++++ .../{ => configs}/upstreams-ethereum-pos.yaml | 0 .../upstreams-method-groups.yaml | 0 .../{ => configs}/upstreams-node-id.yaml | 0 14 files changed, 220 insertions(+), 32 deletions(-) create mode 100644 src/test/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigTest.kt create mode 100644 src/test/resources/configs/upstreams-connector-mode.yaml rename src/test/resources/{ => configs}/upstreams-ethereum-pos.yaml (100%) rename src/test/resources/{ => configs}/upstreams-method-groups.yaml (100%) rename src/test/resources/{ => configs}/upstreams-node-id.yaml (100%) diff --git a/docs/04-upstream-config.adoc b/docs/04-upstream-config.adoc index 03279dc0..9dccc2d5 100644 --- a/docs/04-upstream-config.adoc +++ b/docs/04-upstream-config.adoc @@ -190,6 +190,27 @@ Dshackle currently supports - `ws` websocket connection (supposed to be used in addition to `rpc` connection) - `grpc` connects to another Dshackle instance +==== Connection mixture modes +In case of rpc and ws connection we can specify different modes of works together: + +|=== +|Type |Description + +|WS_ONLY +|Default mode in case WS endpoint specified. In this mode WS connection is used for all requests and subscriptions. + +|RPC_ONLY +|Default in case WS endpoint not specified. In this mode RPC connection is used for all requests, subscriptions doesn't work, head subscription works through scheduled RPC head request. + +|RPC_REQUESTS_WITH_MIXED_HEAD +|All requests are sent through RPC connection, eth_subscribe is sent through WS connection, head subscription works through scheduled RPC head request mixed with WS subscription. + +|RPC_REQUESTS_WITH_WS_HEAD +|All requests are sent through RPC connection, all subscriptions works through WS connection. +|=== + +You can specify this modes through `connector-mode` parameter in connection config. + === Bitcoin Methods .By default an ethereum upstream allows call to the following JSON RPC methods: diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 996dc85c..e14837c9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.config import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode import java.net.URI import java.util.Arrays import java.util.Locale @@ -120,6 +121,23 @@ open class UpstreamsConfig { class EthereumConnection : RpcConnection() { var ws: WsEndpoint? = null var preferHttp: Boolean = false + var connectorMode: String? = null + + fun resolveMode(): ConnectorMode { + return if (preferHttp) { + ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD + } else { + if (connectorMode == null) { + if (ws == null) { + ConnectorMode.RPC_ONLY + } else { + ConnectorMode.WS_ONLY + } + } else { + ConnectorMode.parse(connectorMode!!) + } + } + } } class BitcoinConnection : RpcConnection() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index 134727bb..95c51ebd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -186,6 +186,9 @@ class UpstreamsConfigReader( getValueAsBool(connConfigNode, "prefer-http")?.let { connection.preferHttp = it } + getValueAsString(connConfigNode, "connector-mode")?.let { + connection.connectorMode = it + } getMapping(connConfigNode, "ws")?.let { node -> getValueAsString(node, "url")?.let { url -> val ws = UpstreamsConfig.WsEndpoint(URI(url)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 1886b3be..a008c58e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -105,6 +105,15 @@ open class ConfiguredUpstreams( val options = (defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults()) .merge(up.options ?: UpstreamsConfig.Options()) val upstream = when (BlockchainType.from(chain)) { + BlockchainType.EVM_POS -> { + buildEthereumPosUpstream( + up.nodeId, + up.cast(UpstreamsConfig.EthereumPosConnection::class.java), + chain, + options, + chainsConfig.resolve(chain) + ) + } BlockchainType.EVM_POW -> { buildEthereumUpstream( up.nodeId, @@ -114,7 +123,6 @@ open class ConfiguredUpstreams( chainsConfig.resolve(chain) ) } - BlockchainType.BITCOIN -> { buildBitcoinUpstream( up.cast(UpstreamsConfig.BitcoinConnection::class.java), @@ -123,16 +131,6 @@ open class ConfiguredUpstreams( chainsConfig.resolve(chain) ) } - - BlockchainType.EVM_POS -> { - buildEthereumPosUpstream( - up.nodeId, - up.cast(UpstreamsConfig.EthereumPosConnection::class.java), - chain, - options, - chainsConfig.resolve(chain) - ) - } } upstream?.let { val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED) @@ -211,7 +209,7 @@ open class ConfiguredUpstreams( } val hashUrl = conn.execution!!.let { - if (it.preferHttp) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url + if (it.preferHttp == true) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url } val hash = getHash(nodeId, hashUrl!!) val upstream = EthereumPosRpcUpstream( @@ -295,7 +293,7 @@ open class ConfiguredUpstreams( return null } - val hashUrl = if (conn.preferHttp) conn.rpc?.url ?: conn.ws?.url else conn.ws?.url ?: conn.rpc?.url + val hashUrl = if (conn.preferHttp == true) conn.rpc?.url ?: conn.ws?.url else conn.ws?.url ?: conn.rpc?.url val upstream = EthereumRpcUpstream( config.id!!, getHash(nodeId, hashUrl!!), @@ -399,7 +397,7 @@ open class ConfiguredUpstreams( val httpFactory = buildHttpFactory(conn, urls) log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") val connectorFactory = - EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice, blockValidator) + EthereumConnectorFactory(conn.resolveMode(), wsFactoryApi, httpFactory, forkChoice, blockValidator) if (!connectorFactory.isValid()) { log.warn("Upstream configuration is invalid (probably no http endpoint)") return null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt index 103cb4ac..9f623960 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt @@ -286,7 +286,7 @@ open class WsConnectionImpl( ) val sender = currentRequests.remove(msg.id.asNumber().toInt()) if (sender == null) { - log.warn("Unknown response received for ${msg.id}") + log.warn("Unknown response received for ${msg.id} with body ${msg.value?.let { String(it) }}") } else { try { val emitResult = sender.tryEmitValue(rpcResponse) 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 9cf9b8f9..045280de 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 @@ -6,10 +6,14 @@ import io.emeraldpay.dshackle.upstream.DefaultUpstream 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.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_ONLY +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.WS_ONLY import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice open class EthereumConnectorFactory( - private val preferHttp: Boolean, + private val connectorType: ConnectorMode, private val wsFactory: EthereumWsConnectionPoolFactory?, private val httpFactory: HttpFactory?, private val forkChoice: ForkChoice, @@ -17,7 +21,20 @@ open class EthereumConnectorFactory( ) : ConnectorFactory { override fun isValid(): Boolean { - if (preferHttp && httpFactory == null) { + if (( + connectorType == RPC_ONLY || + connectorType == RPC_REQUESTS_WITH_MIXED_HEAD || + connectorType == RPC_REQUESTS_WITH_WS_HEAD + ) && httpFactory == null + ) { + return false + } + if (( + connectorType == WS_ONLY || + connectorType == RPC_REQUESTS_WITH_MIXED_HEAD || + connectorType == RPC_REQUESTS_WITH_WS_HEAD + ) && wsFactory == null + ) { return false } return true @@ -29,13 +46,14 @@ open class EthereumConnectorFactory( chain: Chain, skipEnhance: Boolean ): EthereumConnector { - if (wsFactory != null && !preferHttp) { + if (wsFactory != null && connectorType == WS_ONLY) { 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( + connectorType, httpFactory.create(upstream.getId(), chain), wsFactory, upstream.getId(), @@ -44,4 +62,22 @@ open class EthereumConnectorFactory( skipEnhance ) } + + enum class ConnectorMode { + WS_ONLY, + RPC_ONLY, + RPC_REQUESTS_WITH_MIXED_HEAD, + RPC_REQUESTS_WITH_WS_HEAD; + + companion object { + val values = values().map { it.name }.toSet() + fun parse(value: String): ConnectorMode { + val upper = value.uppercase() + if (!values.contains(upper)) { + throw IllegalArgumentException("Invalid connector mode: $value") + } + return valueOf(upper) + } + } + } } 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 cfdb0bb0..9421bd43 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 @@ -14,12 +14,18 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead import io.emeraldpay.dshackle.upstream.ethereum.NoEthereumIngressSubscription import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptionsImpl +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_ONLY +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode.WS_ONLY import io.emeraldpay.dshackle.upstream.forkchoice.AlwaysForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import org.slf4j.LoggerFactory import java.time.Duration class EthereumRpcConnector( + connectorType: ConnectorMode, private val directReader: JsonRpcReader, wsFactory: EthereumWsConnectionPoolFactory?, id: String, @@ -36,19 +42,30 @@ class EthereumRpcConnector( init { if (wsFactory != null) { - // 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, 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)) - head = MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id") } else { pool = null - log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC") - head = EthereumRpcHead(getIngressReader(), forkChoice, id, blockValidator) + } + + head = when (connectorType) { + RPC_ONLY -> { + log.warn("Setting up connector for $id upstream with RPC-only access, less effective than WS+RPC") + EthereumRpcHead(getIngressReader(), forkChoice, id, blockValidator) + } + WS_ONLY -> { + throw IllegalStateException("WS-only mode is not supported in RPC connector") + } + RPC_REQUESTS_WITH_MIXED_HEAD -> { + val wsHead = + EthereumWsHead(id, AlwaysForkChoice(), blockValidator, getIngressReader(), WsSubscriptionsImpl(pool!!), 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)) + MergedHead(listOf(rpcHead, wsHead), forkChoice, "Merged for $id") + } + RPC_REQUESTS_WITH_WS_HEAD -> { + EthereumWsHead(id, AlwaysForkChoice(), blockValidator, getIngressReader(), WsSubscriptionsImpl(pool!!), skipEnhance) + } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index a6100728..bf494d15 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -153,7 +153,7 @@ class UpstreamsConfigReaderSpec extends Specification { def "Parse ethereum pos upstreams"() { setup: - def config = this.class.getClassLoader().getResourceAsStream("upstreams-ethereum-pos.yaml") + def config = this.class.getClassLoader().getResourceAsStream("configs/upstreams-ethereum-pos.yaml") when: def act = reader.readInternal(config) then: @@ -404,7 +404,7 @@ class UpstreamsConfigReaderSpec extends Specification { def "Parse node id"() { setup: - def config = this.class.getClassLoader().getResourceAsStream("upstreams-node-id.yaml") + def config = this.class.getClassLoader().getResourceAsStream("configs/upstreams-node-id.yaml") when: def act = reader.readInternal(config) then: @@ -422,7 +422,7 @@ class UpstreamsConfigReaderSpec extends Specification { def "Parse method groups"() { setup: - def config = this.class.getClassLoader().getResourceAsStream("upstreams-method-groups.yaml") + def config = this.class.getClassLoader().getResourceAsStream("configs/upstreams-method-groups.yaml") when: def act = reader.readInternal(config) then: @@ -459,4 +459,26 @@ class UpstreamsConfigReaderSpec extends Specification { validatePeers == true } } + + def "Parse connector mode in connection config"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("configs/upstreams-connector-mode.yaml") + when: + def act = reader.readInternal(config) + then: + act != null + act.upstreams.size() == 1 + with(act.upstreams.get(0)) { + id == "local" + chain == "ethereum" + connection instanceof UpstreamsConfig.EthereumConnection + with((UpstreamsConfig.EthereumConnection) connection) { + rpc != null + rpc.url == new URI("https://localhost:8546") + ws != null + ws.url == new URI("ws://localhost:8546") + connectorMode == "RPC_REQUESTS_WITH_WS_HEAD" + } + } + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 100f72ae..a374aa16 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -49,7 +49,7 @@ class FilteredApisSpec extends Specification { def httpFactory = Mock(HttpFactory) { create(_, _) >> TestingCommons.api().tap { it.id = "${i++}" } } - def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory, new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID) + def connectorFactory = new EthereumConnectorFactory(EthereumConnectorFactory.ConnectorMode.RPC_ONLY, null, httpFactory, new MostWorkForkChoice(), BlockValidator.ALWAYS_VALID) new EthereumRpcUpstream( "test", (byte)123, diff --git a/src/test/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigTest.kt b/src/test/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigTest.kt new file mode 100644 index 00000000..5b7affd8 --- /dev/null +++ b/src/test/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigTest.kt @@ -0,0 +1,61 @@ +package io.emeraldpay.dshackle.config + +import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory.ConnectorMode +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.Arguments +import org.junit.jupiter.params.provider.MethodSource +import java.net.URI +import java.util.stream.Stream + +internal class UpstreamsConfigTest { + companion object { + @JvmStatic + fun data(): Stream { + return Stream.of( + Arguments.of( + UpstreamsConfig.EthereumConnection(), + ConnectorMode.RPC_ONLY + ), + Arguments.of( + UpstreamsConfig.EthereumConnection() + .apply { + ws = UpstreamsConfig.WsEndpoint(URI("ws://localhost:8546")) + }, + ConnectorMode.WS_ONLY + ), + Arguments.of( + UpstreamsConfig.EthereumConnection() + .apply { + preferHttp = true + ws = UpstreamsConfig.WsEndpoint(URI("ws://localhost:8546")) + }, + ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD + ), + Arguments.of( + UpstreamsConfig.EthereumConnection() + .apply { + connectorMode = "RPC_REQUESTS_WITH_WS_HEAD" + ws = UpstreamsConfig.WsEndpoint(URI("ws://localhost:8546")) + }, + ConnectorMode.RPC_REQUESTS_WITH_WS_HEAD + ), + Arguments.of( + UpstreamsConfig.EthereumConnection() + .apply { + preferHttp = true + connectorMode = "RPC_REQUESTS_WITH_WS_HEAD" + ws = UpstreamsConfig.WsEndpoint(URI("ws://localhost:8546")) + }, + ConnectorMode.RPC_REQUESTS_WITH_MIXED_HEAD + ), + ) + } + } + + @ParameterizedTest + @MethodSource("data") + fun testKeepForwarded(input: UpstreamsConfig.EthereumConnection, expected: ConnectorMode) { + assertEquals(expected, input.resolveMode()) + } +} diff --git a/src/test/resources/configs/upstreams-connector-mode.yaml b/src/test/resources/configs/upstreams-connector-mode.yaml new file mode 100644 index 00000000..a35f0abd --- /dev/null +++ b/src/test/resources/configs/upstreams-connector-mode.yaml @@ -0,0 +1,12 @@ +version: v1 + +upstreams: + - id: local + chain: ethereum + connection: + ethereum: + connector-mode: RPC_REQUESTS_WITH_WS_HEAD + ws: + url: "ws://localhost:8546" + rpc: + url: "https://localhost:8546" \ No newline at end of file diff --git a/src/test/resources/upstreams-ethereum-pos.yaml b/src/test/resources/configs/upstreams-ethereum-pos.yaml similarity index 100% rename from src/test/resources/upstreams-ethereum-pos.yaml rename to src/test/resources/configs/upstreams-ethereum-pos.yaml diff --git a/src/test/resources/upstreams-method-groups.yaml b/src/test/resources/configs/upstreams-method-groups.yaml similarity index 100% rename from src/test/resources/upstreams-method-groups.yaml rename to src/test/resources/configs/upstreams-method-groups.yaml diff --git a/src/test/resources/upstreams-node-id.yaml b/src/test/resources/configs/upstreams-node-id.yaml similarity index 100% rename from src/test/resources/upstreams-node-id.yaml rename to src/test/resources/configs/upstreams-node-id.yaml