problem: WS connection still breaks with very large frames

solution: increase default limit and make it configurable

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

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

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 {

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 {

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
             }

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,
+                                    .maxFramePayloadLength(frameSize)
-                                    // 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)
                                     .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(msgSizeLimit)
-                    .aggregateFrames(15 * 1024 * 1024)
                     .receiveFrames()
                     .map { ByteBufInputStream(it.content()).readAllBytes() }
                     .flatMap {

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:

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

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