Add ws connection pool

2023-01-26 15:53:00 +04:00
parent 731ee2c747
commit ed0c5cb517
24 changed files with 576 additions and 257 deletions
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
@@ -137,6 +137,7 @@ open class UpstreamsConfig {
        var basicAuth: AuthConfig.ClientBasicAuth? = null
        var frameSize: Int? = null
        var msgSize: Int? = null
+        var connections: Int = 1
    }

    // TODO make it unmodifiable after initial load
--- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
@@ -126,14 +126,8 @@ class UpstreamsConfigReader(

    private fun readBitcoinConnection(connConfigNode: MappingNode): UpstreamsConfig.BitcoinConnection {
        val connection = UpstreamsConfig.BitcoinConnection()
-        getMapping(connConfigNode, "rpc")?.let { node ->
-            getValueAsString(node, "url")?.let { url ->
-                val http = UpstreamsConfig.HttpEndpoint(URI(url))
-                connection.rpc = http
-                http.basicAuth = authConfigReader.readClientBasicAuth(node)
-                http.tls = authConfigReader.readClientTls(node)
-            }
-        }
+            .apply { rpc = readRpcConfig(connConfigNode) }
+
        getMapping(connConfigNode, "esplora")?.let { node ->
            getValueAsString(node, "url")?.let { url ->
                val http = UpstreamsConfig.HttpEndpoint(URI(url))
@@ -164,6 +158,17 @@ class UpstreamsConfigReader(
        return connection
    }

+    private fun readRpcConfig(connConfigNode: MappingNode): UpstreamsConfig.HttpEndpoint? {
+        return getMapping(connConfigNode, "rpc")?.let { node ->
+            getValueAsString(node, "url")?.let { url ->
+                val http = UpstreamsConfig.HttpEndpoint(URI(url))
+                http.basicAuth = authConfigReader.readClientBasicAuth(node)
+                http.tls = authConfigReader.readClientTls(node)
+                http
+            }
+        }
+    }
+
    private fun readEthereumPosConnection(connConfigNode: MappingNode): UpstreamsConfig.EthereumPosConnection {
        val connection = UpstreamsConfig.EthereumPosConnection()
        getMapping(connConfigNode, "execution")?.let {
@@ -174,16 +179,11 @@ class UpstreamsConfigReader(
        }
        return connection
    }
+
    private fun readEthereumConnection(connConfigNode: MappingNode): UpstreamsConfig.EthereumConnection {
        val connection = UpstreamsConfig.EthereumConnection()
-        getMapping(connConfigNode, "rpc")?.let { node ->
-            getValueAsString(node, "url")?.let { url ->
-                val http = UpstreamsConfig.HttpEndpoint(URI(url))
-                connection.rpc = http
-                http.basicAuth = authConfigReader.readClientBasicAuth(node)
-                http.tls = authConfigReader.readClientTls(node)
-            }
-        }
+            .apply { rpc = readRpcConfig(connConfigNode) }
+
        getValueAsBool(connConfigNode, "prefer-http")?.let {
            connection.preferHttp = it
        }
@@ -208,6 +208,12 @@ class UpstreamsConfigReader(
                    }
                    ws.msgSize = it
                }
+                getValueAsInt(node, "connections")?.let {
+                    if (it < 1 || it > 1024) {
+                        throw IllegalStateException("connection limit should be in 1..1024")
+                    }
+                    ws.connections = it
+                }
            }
        }
        return connection
--- a/src/main/kotlin/io/emeraldpay/dshackle/data/RawJsonBuilder.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/data/RawJsonBuilder.kt
@@ -1,47 +0,0 @@
-/**
- * Copyright (c) 2020 EmeraldPay, Inc
- * Copyright (c) 2020 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.data
-
-import org.slf4j.LoggerFactory
-import java.io.ByteArrayOutputStream
-
-class RawJsonBuilder {
-
-    companion object {
-        private val log = LoggerFactory.getLogger(RawJsonBuilder::class.java)
-
-        private val START = "{\"jsonrpc\":\"2.0\"".toByteArray()
-        private val ID_START = "\"id\":".toByteArray()
-        private val RESULT_START = "\"result\":".toByteArray()
-        private val COMMA = ",".toByteArray()
-        private val END = "}".toByteArray()
-    }
-
-    fun write(id: Int, data: ByteArray): ByteArray {
-        val buf = ByteArrayOutputStream(data.size + 100)
-        buf.write(START)
-        buf.write(COMMA)
-        buf.write(ID_START)
-        buf.write(id.toString().toByteArray())
-        buf.write(COMMA)
-        buf.write(RESULT_START)
-        buf.write(data)
-        buf.write(END)
-
-        return buf.toByteArray()
-    }
-}
--- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
@@ -35,7 +35,8 @@ import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumBlockValidator
 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.EthereumWsConnectionFactory
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory
 import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice
@@ -337,17 +338,21 @@ open class ConfiguredUpstreams(
        chain: Chain,
        conn: UpstreamsConfig.EthereumConnection,
        urls: ArrayList<URI>? = null
-    ): EthereumWsFactory? {
+    ): EthereumWsConnectionPoolFactory? {
        return conn.ws?.let { endpoint ->
-            val wsApi = EthereumWsFactory(
+            val wsConnectionFactory = EthereumWsConnectionFactory(
                id, chain,
                endpoint.url,
                endpoint.origin ?: URI("http://localhost"),
-            )
-            wsApi.config = endpoint
-            endpoint.basicAuth?.let { auth ->
-                wsApi.basicAuth = auth
+            ).apply {
+                config = endpoint
+                basicAuth = endpoint.basicAuth
            }
+            val wsApi = EthereumWsConnectionPoolFactory(
+                id,
+                endpoint.connections,
+                wsConnectionFactory
+            )
            urls?.add(endpoint.url)
            wsApi
        }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionFactory.kt
@@ -1,25 +1,8 @@
-/**
- * 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.Chain
 import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
-import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 import io.micrometer.core.instrument.Counter
 import io.micrometer.core.instrument.Metrics
@@ -27,7 +10,7 @@ import io.micrometer.core.instrument.Tag
 import io.micrometer.core.instrument.Timer
 import java.net.URI

-class EthereumWsFactory(
+open class EthereumWsConnectionFactory(
    private val id: String,
    private val chain: Chain,
    private val uri: URI,
@@ -37,15 +20,15 @@ class EthereumWsFactory(
    var basicAuth: AuthConfig.ClientBasicAuth? = null
    var config: UpstreamsConfig.WsEndpoint? = null

-    // metrics are shared between all connections to the same WS
-    private val metrics: RpcMetrics = run {
+    private fun metrics(connIndex: Int): RpcMetrics {
        val metricsTags = listOf(
+            Tag.of("index", connIndex.toString()),
            Tag.of("upstream", id),
            // UNSPECIFIED shouldn't happen too
            Tag.of("chain", chain.chainCode)
        )

-        RpcMetrics(
+        return RpcMetrics(
            Timer.builder("upstream.ws.conn")
                .description("Request time through a WebSocket JSON RPC connection")
                .tags(metricsTags)
@@ -58,11 +41,8 @@ class EthereumWsFactory(
        )
    }

-    fun create(upstream: DefaultUpstream?): WsConnectionImpl {
-        require(upstream == null || upstream.getId() == id) {
-            "Creating instance for different upstream. ${upstream?.getId()} != id"
-        }
-        return WsConnectionImpl(id, uri, origin, basicAuth, metrics, upstream).also { ws ->
+    open fun createWsConnection(connIndex: Int = 0, onDisconnect: () -> Unit): WsConnection =
+        WsConnectionImpl(uri, origin, basicAuth, metrics(connIndex), onDisconnect).also { ws ->
            config?.frameSize?.let {
                ws.frameSize = it
            }
@@ -70,5 +50,4 @@ class EthereumWsFactory(
                ws.msgSizeLimit = it
            }
        }
-    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionPoolFactory.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsConnectionPoolFactory.kt
@@ -0,0 +1,37 @@
+/**
+ * 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.upstream.DefaultUpstream
+
+class EthereumWsConnectionPoolFactory(
+    private val id: String,
+    private val connections: Int,
+    private val ethereumWsConnectionFactory: EthereumWsConnectionFactory
+) {
+
+    fun create(upstream: DefaultUpstream?): WsConnectionPool {
+        require(upstream == null || upstream.getId() == id) {
+            "Creating instance for different upstream. ${upstream?.getId()} != id"
+        }
+        return if (connections > 1) {
+            WsConnectionMultiPool(ethereumWsConnectionFactory, upstream, connections)
+        } else {
+            WsConnectionSinglePool(ethereumWsConnectionFactory, upstream)
+        }
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt
@@ -0,0 +1,31 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * 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.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsMessage
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+interface WsConnection : AutoCloseable {
+
+    val isConnected: Boolean
+
+    fun getSubscribeResponses(): Flux<JsonRpcWsMessage>
+    fun callRpc(originalRequest: JsonRpcRequest): Mono<JsonRpcResponse>
+    fun connect()
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionImpl.kt
@@ -18,8 +18,6 @@ package io.emeraldpay.dshackle.upstream.ethereum
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.config.AuthConfig
-import io.emeraldpay.dshackle.upstream.DefaultUpstream
-import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -28,6 +26,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsMessage
 import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser
 import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 import io.emeraldpay.etherjar.rpc.RpcResponseError
+import io.micrometer.core.instrument.Metrics
 import io.netty.buffer.ByteBuf
 import io.netty.buffer.ByteBufInputStream
 import io.netty.buffer.Unpooled
@@ -61,13 +60,12 @@ import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger

 open class WsConnectionImpl(
-    private val id: String,
    private val uri: URI,
    private val origin: URI,
    private val basicAuth: AuthConfig.ClientBasicAuth?,
    private val rpcMetrics: RpcMetrics?,
-    private val upstream: DefaultUpstream?,
-) : AutoCloseable {
+    private val onDisconnect: () -> Unit
+) : AutoCloseable, WsConnection, Cloneable {

    companion object {
        private val log = LoggerFactory.getLogger(WsConnectionImpl::class.java)
@@ -124,7 +122,7 @@ open class WsConnectionImpl(
    private var connection: Disposable? = null
    private val reconnecting = AtomicBoolean(false)

-    open val isConnected: Boolean
+    override val isConnected: Boolean
        get() = connection != null && !reconnecting.get()

    fun setReconnectIntervalSeconds(value: Long) {
@@ -132,7 +130,7 @@ open class WsConnectionImpl(
        currentBackOff = reconnectBackoff.start()
    }

-    fun connect() {
+    override fun connect() {
        keepConnection = true
        connectInternal()
    }
@@ -179,10 +177,9 @@ open class WsConnectionImpl(
        connection = HttpClient.create()
            .resolver(DefaultAddressResolverGroup.INSTANCE)
            .doOnDisconnected {
+                onDisconnect()
                disconnects.tryEmitNext(Instant.now())
                log.info("Disconnected from $uri")
-                // mark upstream as UNAVAIL
-                upstream?.setStatus(UpstreamAvailability.UNAVAILABLE)
                if (keepConnection) {
                    tryReconnectLater()
                }
@@ -274,7 +271,7 @@ open class WsConnectionImpl(
        )
    }

-    fun onMessage(msg: ResponseWSParser.WsResponse): Mono<Void> {
+    private fun onMessage(msg: ResponseWSParser.WsResponse): Mono<Void> {
        return Mono.fromCallable {
            when (msg.type) {
                ResponseWSParser.Type.RPC -> onMessageRpc(msg)
@@ -283,7 +280,7 @@ open class WsConnectionImpl(
        }.then()
    }

-    fun onMessageRpc(msg: ResponseWSParser.WsResponse) {
+    private fun onMessageRpc(msg: ResponseWSParser.WsResponse) {
        val rpcResponse = JsonRpcResponse(
            msg.value, msg.error, msg.id, null
        )
@@ -302,7 +299,7 @@ open class WsConnectionImpl(
        }
    }

-    fun onMessageSubscription(msg: ResponseWSParser.WsResponse) {
+    private fun onMessageSubscription(msg: ResponseWSParser.WsResponse) {
        val subscription = JsonRpcWsMessage(
            msg.value, msg.error, msg.id.asString(),
        )
@@ -316,11 +313,11 @@ open class WsConnectionImpl(
        }
    }

-    open fun getSubscribeResponses(): Flux<JsonRpcWsMessage> {
+    override fun getSubscribeResponses(): Flux<JsonRpcWsMessage> {
        return Flux.from(subscriptionResponses.asFlux())
    }

-    open fun callRpc(originalRequest: JsonRpcRequest): Mono<JsonRpcResponse> {
+    override fun callRpc(originalRequest: JsonRpcRequest): Mono<JsonRpcResponse> {
        return Mono.fromCallable {
            val startTime = System.nanoTime()
            // use an internal id sequence, to avoid id conflicts with user calls
@@ -342,7 +339,7 @@ open class WsConnectionImpl(
        }
    }

-    fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono<JsonRpcResponse> {
+    private fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono<JsonRpcResponse> {
        val internalId = request.id.toLong()
        val onResponse = Sinks.one<JsonRpcResponse>()
        currentRequests[internalId.toInt()] = onResponse
@@ -386,5 +383,14 @@ open class WsConnectionImpl(
        connection?.dispose()
        connection = null
        currentRequests.clear()
+        rpcMetrics?.fails?.let {
+            it.close()
+            Metrics.globalRegistry.remove(it)
+        }
+        rpcMetrics?.timer?.let {
+            it.close()
+            Metrics.globalRegistry.remove(it)
+        }
+        resetBackoffExecutor.shutdown()
    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionMultiPool.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionMultiPool.kt
@@ -0,0 +1,133 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * 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.Global
+import io.emeraldpay.dshackle.upstream.DefaultUpstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+import org.springframework.util.backoff.BackOffExecution
+import org.springframework.util.backoff.ExponentialBackOff
+import java.time.Duration
+import java.util.concurrent.ScheduledExecutorService
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.locks.ReentrantReadWriteLock
+import kotlin.concurrent.read
+import kotlin.concurrent.write
+
+/**
+ * A Websocket connection pool which keeps up to `target` connection providing them in a round-robin fashion.
+ *
+ * It doesn't make all the connections immediately but rather grows them one by one ensuring existing are ok.
+ * By default, it adds a new connection every 5 second until it reaches the target number.
+ */
+class WsConnectionMultiPool(
+    private val ethereumWsConnectionFactory: EthereumWsConnectionFactory,
+    private val upstream: DefaultUpstream?,
+    private val connections: Int,
+) : WsConnectionPool {
+
+    companion object {
+        private const val SCHEDULE_FULL = 60L
+        private const val SCHEDULE_GROW = 5L
+        private const val SCHEDULE_BROKEN = 15L
+    }
+
+    private val current = ArrayList<WsConnection>()
+    private var adjustLock = ReentrantReadWriteLock()
+    private val index = AtomicInteger(0)
+    private var connIndex = 0
+
+    var scheduler: ScheduledExecutorService = Global.control
+
+    override fun connect() {
+        adjust()
+    }
+
+    override fun getConnection(): WsConnection {
+        val tries = ExponentialBackOff(50, 1.25).also {
+            it.maxElapsedTime = Duration.ofMinutes(1).toMillis()
+        }.start()
+        var next = next()
+        while (next == null) {
+            val sleep = tries.nextBackOff()
+            if (sleep == BackOffExecution.STOP) {
+                throw IllegalStateException("No available WS connection")
+            }
+            Thread.sleep(sleep)
+            next = next()
+        }
+        return next
+    }
+
+    override fun close() {
+        adjustLock.write {
+            current.forEach { it.close() }
+            current.clear()
+        }
+    }
+
+    private fun next(): WsConnection? {
+        adjustLock.read {
+            if (current.isEmpty()) {
+                return null
+            }
+            return current[index.getAndIncrement() % current.size]
+        }
+    }
+
+    private fun adjust() {
+        adjustLock.write {
+            // add a new connection only if all existing are active or there are no connections at all
+            val allOk = current.all { it.isConnected }
+            val schedule: Long
+            if (allOk) {
+                schedule = if (current.size >= connections) {
+                    // recheck the state in a minute and adjust if any connection went bad
+                    SCHEDULE_FULL
+                } else {
+                    current.add(
+                        ethereumWsConnectionFactory.createWsConnection(connIndex++) {
+                            if (isUnavailable()) {
+                                upstream?.setStatus(UpstreamAvailability.UNAVAILABLE)
+                            }
+                        }.also {
+                            it.connect()
+                        }
+                    )
+                    SCHEDULE_GROW
+                }
+            } else {
+                // technically there is no reason to disconnect because it supposed to reconnect,
+                // but to ensure clean start (or other internal state issues) lets completely close all broken and create new
+                current.removeIf {
+                    if (!it.isConnected) {
+                        // DO NOT FORGET to close the connection, otherwise it would keep reconnecting but unused
+                        it.close()
+                        true
+                    } else {
+                        false
+                    }
+                }
+                schedule = SCHEDULE_BROKEN
+            }
+
+            scheduler.schedule({ adjust() }, schedule, TimeUnit.SECONDS)
+        }
+    }
+
+    private fun isUnavailable() = adjustLock.read { current.count { it.isConnected } == 0 }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionPool.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionPool.kt
@@ -0,0 +1,21 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * 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
+
+interface WsConnectionPool : AutoCloseable {
+    fun connect()
+    fun getConnection(): WsConnection
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSinglePool.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSinglePool.kt
@@ -0,0 +1,42 @@
+/**
+ * Copyright (c) 2022 EmeraldPay, Inc
+ *
+ * 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.upstream.DefaultUpstream
+import io.emeraldpay.dshackle.upstream.UpstreamAvailability
+
+class WsConnectionSinglePool(
+    ethereumWsConnectionFactory: EthereumWsConnectionFactory,
+    private val upstream: DefaultUpstream?,
+) : WsConnectionPool {
+    private val connection = ethereumWsConnectionFactory.createWsConnection {
+        upstream?.setStatus(UpstreamAvailability.UNAVAILABLE)
+    }
+
+    override fun connect() {
+        if (!connection.isConnected) {
+            connection.connect()
+        }
+    }
+
+    override fun getConnection(): WsConnection {
+        return connection
+    }
+
+    override fun close() {
+        connection.close()
+    }
+}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsSubscriptionsImpl.kt
@@ -24,7 +24,7 @@ import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.atomic.AtomicReference

 class WsSubscriptionsImpl(
-    val conn: WsConnectionImpl,
+    val wsPool: WsConnectionPool,
 ) : WsSubscriptions {

    companion object {
@@ -35,6 +35,7 @@ class WsSubscriptionsImpl(

    override fun subscribe(method: String): Flux<ByteArray> {
        val subscriptionId = AtomicReference("")
+        val conn = wsPool.getConnection()
        val messages = conn.getSubscribeResponses()
            .filter { it.subscriptionId == subscriptionId.get() }
            .filter { it.result != null } // should never happen
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumConnectorFactory.kt
@@ -5,13 +5,13 @@ import io.emeraldpay.dshackle.upstream.BlockValidator
 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.EthereumWsFactory
+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: EthereumWsFactory?,
+    private val wsFactory: EthereumWsConnectionPoolFactory?,
    private val httpFactory: HttpFactory?,
    private val forkChoice: ForkChoice,
    private val blockValidator: BlockValidator
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumRpcConnector.kt
@@ -7,7 +7,13 @@ import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.MergedHead
-import io.emeraldpay.dshackle.upstream.ethereum.*
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcHead
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory
+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.forkchoice.AlwaysForkChoice
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import org.slf4j.LoggerFactory
@@ -15,12 +21,12 @@ import java.time.Duration

 class EthereumRpcConnector(
    private val directReader: JsonRpcReader,
-    wsFactory: EthereumWsFactory?,
+    wsFactory: EthereumWsConnectionPoolFactory?,
    id: String,
    forkChoice: ForkChoice,
    blockValidator: BlockValidator
 ) : EthereumConnector, CachesEnabled {
-    private val conn: WsConnectionImpl?
+    private val pool: WsConnectionPool?
    private val head: Head

    companion object {
@@ -30,14 +36,14 @@ class EthereumRpcConnector(
    init {
        if (wsFactory != null) {
            // do not set upstream to the WS, since it doesn't control the RPC upstream
-            conn = wsFactory.create(null)
-            val subscriptions = WsSubscriptionsImpl(conn)
+            pool = wsFactory.create(null)
+            val subscriptions = WsSubscriptionsImpl(pool)
            val wsHead = EthereumWsHead(id, AlwaysForkChoice(), blockValidator, getIngressReader(), subscriptions)
            // 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 {
-            conn = null
+            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)
        }
@@ -50,7 +56,7 @@ class EthereumRpcConnector(
    }

    override fun start() {
-        conn?.connect()
+        pool?.connect()
        if (head is Lifecycle) {
            head.start()
        }
@@ -67,7 +73,7 @@ class EthereumRpcConnector(
        if (head is Lifecycle) {
            head.stop()
        }
-        conn?.close()
+        pool?.close()
    }

    override fun getIngressReader(): JsonRpcReader {
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
@@ -4,32 +4,36 @@ import io.emeraldpay.dshackle.reader.JsonRpcReader
 import io.emeraldpay.dshackle.upstream.BlockValidator
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.Head
-import io.emeraldpay.dshackle.upstream.ethereum.*
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsHead
+import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool
+import io.emeraldpay.dshackle.upstream.ethereum.WsSubscriptionsImpl
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumWsIngressSubscription
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient

 class EthereumWsConnector(
-    wsFactory: EthereumWsFactory,
+    wsFactory: EthereumWsConnectionPoolFactory,
    upstream: DefaultUpstream,
    forkChoice: ForkChoice,
    blockValidator: BlockValidator
 ) : EthereumConnector {
-    private val conn: WsConnectionImpl
+    private val pool: WsConnectionPool
    private val reader: JsonRpcReader
    private val head: EthereumWsHead
    private val subscriptions: EthereumIngressSubscription

    init {
-        conn = wsFactory.create(upstream)
-        reader = JsonRpcWsClient(conn)
-        val wsSubscriptions = WsSubscriptionsImpl(conn)
+        pool = wsFactory.create(upstream)
+        reader = JsonRpcWsClient(pool)
+        val wsSubscriptions = WsSubscriptionsImpl(pool)
        head = EthereumWsHead(upstream.getId(), forkChoice, blockValidator, reader, wsSubscriptions)
        subscriptions = EthereumWsIngressSubscription(wsSubscriptions)
    }

    override fun start() {
-        conn.connect()
+        pool.connect()
        head.start()
    }

@@ -38,7 +42,7 @@ class EthereumWsConnector(
    }

    override fun stop() {
-        conn.close()
+        pool.close()
        head.stop()
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt
@@ -1,27 +0,0 @@
-package io.emeraldpay.dshackle.upstream.rpcclient
-
-import io.emeraldpay.dshackle.reader.JsonRpcReader
-import org.slf4j.LoggerFactory
-import reactor.core.publisher.Mono
-
-/**
- * An aggregating JSON RPC Client that wraps two actual readers, a Primary and a Secondary.
- * It always calls the Primary reader, and if it fails or produces an empty result, then it calls the Secondary reader.
- */
-class JsonRpcSwitchClient(
-    private val primary: JsonRpcReader,
-    private val secondary: JsonRpcReader,
-) : JsonRpcReader {
-
-    companion object {
-        private val log = LoggerFactory.getLogger(JsonRpcSwitchClient::class.java)
-    }
-
-    override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
-        return primary.read(key)
-            .switchIfEmpty(Mono.error(IllegalStateException("No response from Primary Connection")))
-            .onErrorResume {
-                secondary.read(key)
-            }
-    }
-}
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt
@@ -16,16 +16,17 @@
 package io.emeraldpay.dshackle.upstream.rpcclient

 import io.emeraldpay.dshackle.reader.JsonRpcReader
-import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionImpl
+import io.emeraldpay.dshackle.upstream.ethereum.WsConnectionPool
 import io.emeraldpay.etherjar.rpc.RpcResponseError
 import reactor.core.publisher.Mono

 class JsonRpcWsClient(
-    private val ws: WsConnectionImpl
+    private val wsPool: WsConnectionPool
 ) : JsonRpcReader {

    override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
-        if (!ws.isConnected) {
+        val conn = wsPool.getConnection()
+        if (!conn.isConnected) {
            return Mono.error(
                JsonRpcException(
                    JsonRpcResponse.NumberId(key.id),
@@ -36,6 +37,6 @@ class JsonRpcWsClient(
                )
            )
        }
-        return ws.callRpc(key)
+        return conn.callRpc(key)
    }
 }