From ebef8cf27ac63c9743c550a464f3719b2ef30fd8 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 22 Sep 2021 19:35:14 -0400 Subject: [PATCH] problem: WS connection still breaks with very large frames solution: increase default limit and make it configurable --- docs/reference-configuration.adoc | 16 ++++++- .../dshackle/config/UpstreamsConfig.kt | 3 +- .../dshackle/config/UpstreamsConfigReader.kt | 13 ++++++ .../dshackle/config/YamlConfigReader.kt | 16 +++++++ .../dshackle/startup/ConfiguredUpstreams.kt | 3 +- .../upstream/ethereum/EthereumWsFactory.kt | 34 ++++++++++---- .../config/UpstreamsConfigReaderSpec.groovy | 26 +++++++++++ .../config/YamlConfigReaderSpec.groovy | 45 +++++++++++++++++++ src/test/resources/upstreams-ws-full.yaml | 15 +++++++ 9 files changed, 159 insertions(+), 12 deletions(-) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/config/YamlConfigReaderSpec.groovy create mode 100644 src/test/resources/upstreams-ws-full.yaml diff --git a/docs/reference-configuration.adoc b/docs/reference-configuration.adoc index 49df1a79..b5b9aea3 100644 --- a/docs/reference-configuration.adoc +++ b/docs/reference-configuration.adoc @@ -527,6 +527,8 @@ configuration, and may be omitted for most of the situations. basic-auth: username: 9c199ad8f281f20154fc258fe41a6814 password: 258fe4149c199ad8f2811a68f20154fc + frameSize: 5mb + msgSize: 15mb ---- .Main Config @@ -592,7 +594,7 @@ Example: `https://kovan.infura.io/v3/${INFURA_USER}` | `rpc.basic-auth` + `rpc.basic-auth.username`, `rpc.basic-auth.password` a| HTTP Basic Auth configuration, if required by the remote server. + - Values can also reference env variables, for example: +Values can also reference env variables, for example: [source,yaml] ---- rpc: @@ -603,7 +605,8 @@ rpc: ---- | `ws.url` -| Websocket URL to connect to. Optional, but optimizes performance if it's available. +| Websocket URL to connect to. +Optional, but optimizes performance if it's available. | `ws.origin` | HTTP `Origin` if required by Websocket remote server. @@ -611,6 +614,15 @@ rpc: | `ws.basic-auth` + ... | Websocket Basic Auth configuration, if required by the remote server +| `ws.frameSize` +| WebSocket frame size limit. +Ex `1kb`, `1024` (same as `1kb), `2mb`, etc. +Default is 5Mb + +| `ws.msgSize` +| Total limit for a message size consisting from multiple frames. +Ex `1kb`, `1024` (same as `1kb), `2mb`, etc. +Default is 15Mb |=== diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index 4be0ba84..3567fe18 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -121,10 +121,11 @@ open class UpstreamsConfig { class WsEndpoint(val url: URI) { var origin: URI? = null var basicAuth: AuthConfig.ClientBasicAuth? = null + var frameSize: Int? = null + var msgSize: Int? = null } - //TODO make it unmodifiable after initial load class Labels: HashMap() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt index b9f42934..9f6e7cbe 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt @@ -112,6 +112,19 @@ class UpstreamsConfigReader( ws.origin = URI(origin) } ws.basicAuth = authConfigReader.readClientBasicAuth(node) + + getValueAsBytes(node, "frameSize")?.let { + if (it < 65_535) { + throw IllegalStateException("frameSize cannot be less than 64Kb") + } + ws.frameSize = it + } + getValueAsBytes(node, "msgSize")?.let { + if (it < 65_535) { + throw IllegalStateException("msgSize cannot be less than 64Kb") + } + ws.msgSize = it + } } } } else { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt index fcd432e8..ea508a74 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt @@ -127,6 +127,22 @@ abstract class YamlConfigReader { } } + fun getValueAsBytes(mappingNode: MappingNode?, key: String): Int? { + return getValueAsString(mappingNode, key)?.let(envVariables::postProcess)?.let { + val m = Regex("^(\\d+)(m|mb|k|kb|b)?$").find(it.lowercase().trim()) + ?: throw IllegalArgumentException("Not a data size: ${it}. Example of correct values: '1024', '1kb', '5mb'") + val multiplier = m.groups[2]?.let { + when (it.value) { + "k", "kb" -> 1024 + "m", "mb" -> 1024 * 1024 + else -> 1 + } + } ?: 1 + val base = m.groups[1]!!.value.toInt() + base * multiplier + } + } + // ---- fun getBlockchain(id: String): Chain { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 81e0d55f..f2042964 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -195,8 +195,9 @@ open class ConfiguredUpstreams( val wsFactoryApi: EthereumWsFactory? = conn.ws?.let { endpoint -> val wsApi = EthereumWsFactory( endpoint.url, - endpoint.origin ?: URI("http://localhost") + endpoint.origin ?: URI("http://localhost"), ) + wsApi.config = endpoint endpoint.basicAuth?.let { auth -> wsApi.basicAuth = auth } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt index 60b98efc..e2f70168 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.Defaults import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.config.AuthConfig +import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability @@ -65,9 +66,17 @@ class EthereumWsFactory( ) { var basicAuth: AuthConfig.ClientBasicAuth? = null + var config: UpstreamsConfig.WsEndpoint? = null fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?, rpcMetrics: RpcMetrics?): EthereumWs { - return EthereumWs(uri, origin, basicAuth, rpcMetrics, upstream, validator) + return EthereumWs(uri, origin, basicAuth, rpcMetrics, upstream, validator).also { ws -> + config?.frameSize?.let { + ws.frameSize = it + } + config?.msgSize?.let { + ws.msgSizeLimit = it + } + } } class EthereumWs( @@ -84,8 +93,22 @@ class EthereumWsFactory( private const val IDS_START = 100 private const val START_REQUEST = "{\"jsonrpc\":\"2.0\", \"method\":\"eth_subscribe\", \"id\":\"blocks\", \"params\":[\"newHeads\"]}" + + // WebSocket Frame limit. + // Default is 65_536, but Geth responds with larger frames, + // and connection gets dropped with: + // > io.netty.handler.codec.http.websocketx.CorruptedWebSocketFrameException: Max frame length of 65536 has been exceeded + // It's unclear what is a right limit here, but 5mb seems to be working (1mb wasn't always working) + private const val DEFAULT_FRAME_SIZE = 5 * 1024 * 1024 + + // The max size from multiple frames that may represent a single message + // Accept up to 15Mb messages, because Geth is using 15mb, though it's not clear what it limits + private const val DEFAULT_MSG_SIZE = 15 * 1024 * 1024 } + var frameSize: Int = DEFAULT_FRAME_SIZE + var msgSizeLimit: Int = DEFAULT_MSG_SIZE + private var reconnectBackoff: BackOff = ExponentialBackOff().also { it.initialInterval = Duration.ofMillis(100).toMillis() it.maxInterval = Duration.ofMinutes(1).toMillis() @@ -187,11 +210,7 @@ class EthereumWsFactory( WebsocketClientSpec.builder() .handlePing(true) .compress(false) - // Default is 65_536, but Geth responds with larger frames, - // and connection gets dropped with: - // > io.netty.handler.codec.http.websocketx.CorruptedWebSocketFrameException: Max frame length of 65536 has been exceeded - // It's unclear what is a right limit here, but 1mb seems to be working - .maxFramePayloadLength(1024 * 1024) + .maxFramePayloadLength(frameSize) .build() ) .uri(uri) @@ -213,8 +232,7 @@ class EthereumWsFactory( validator?.validate() val consumer = inbound - // Accept up to 15Mb messages, same config is used by Geth - .aggregateFrames(15 * 1024 * 1024) + .aggregateFrames(msgSizeLimit) .receiveFrames() .map { ByteBufInputStream(it.content()).readAllBytes() } .flatMap { diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 54608f81..61bf80f9 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -99,6 +99,32 @@ class UpstreamsConfigReaderSpec extends Specification { } } + def "Parse full defined websocket config"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-ws-full.yaml") + when: + def act = reader.read(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 + ws != null + ws.url == new URI("ws://localhost:8546") + ws.basicAuth != null + with(ws.basicAuth) { + username == "9c199ad8f281f20154fc258fe41a6814" + password == "258fe4149c199ad8f2811a68f20154fc" + } + ws.frameSize == 10 * 1024 * 1024 + ws.msgSize == 25 * 1024 * 1024 + } + } + } def "Parse bitcoin upstreams"() { setup: diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/YamlConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/YamlConfigReaderSpec.groovy new file mode 100644 index 00000000..8e7cd277 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/config/YamlConfigReaderSpec.groovy @@ -0,0 +1,45 @@ +/** + * Copyright (c) 2021 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.config + +import org.yaml.snakeyaml.Yaml +import org.yaml.snakeyaml.nodes.MappingNode +import spock.lang.Specification + +class YamlConfigReaderSpec extends Specification { + + def "reads bytes values"() { + setup: + def rdr = new Impl() + expect: + rdr.getValueAsBytes(asNode("test", input), "test") == exp + where: + input | exp + "1024" | 1024 + "1k" | 1024 + "1kb" | 1024 + "1K" | 1024 + "16kb" | 16 * 1024 + "1M" | 1024 * 1024 + "4mb" | 4 * 1024 * 1024 + } + + private MappingNode asNode(String key, String value) { + return new Yaml().compose(new StringReader("$key: $value")) as MappingNode + } + + class Impl extends YamlConfigReader {} +} diff --git a/src/test/resources/upstreams-ws-full.yaml b/src/test/resources/upstreams-ws-full.yaml new file mode 100644 index 00000000..2d9261f0 --- /dev/null +++ b/src/test/resources/upstreams-ws-full.yaml @@ -0,0 +1,15 @@ +version: v1 + +upstreams: + - id: local + chain: ethereum + connection: + ethereum: + ws: + url: "ws://localhost:8546" + origin: "http://localhost" + frameSize: 10Mb + msgSize: 25Mb + basic-auth: + username: 9c199ad8f281f20154fc258fe41a6814 + password: 258fe4149c199ad8f2811a68f20154fc \ No newline at end of file