Upstream configuration improvement: (#149)

- now we can specify connector mode for better fine tuning of connection

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() {

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))

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

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)

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)
+            }
+        }
+    }
 }

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)
+            }
         }
     }
 

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"
+            }
+        }
+    }
 }

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,

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<Arguments> {
+            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())
+    }
+}

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"