From 568a045271730ac2acc769f2ec7a84d620fba990 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 19 Sep 2021 22:52:21 -0400 Subject: [PATCH 01/12] solution: Ethereum WS connection fetches block through the same WS channel, instead of HTTP RPC --- .../kotlin/io/emeraldpay/dshackle/Global.kt | 2 + .../upstream/ethereum/EthereumRpcUpstream.kt | 2 +- .../upstream/ethereum/EthereumWsFactory.kt | 222 +++++++++++------ .../upstream/ethereum/EthereumWsHead.kt | 2 +- .../upstream/ethereum/NativeCallRouter.kt | 6 +- .../upstream/rpcclient/JsonRpcError.kt | 4 + .../upstream/rpcclient/JsonRpcHttpClient.kt | 4 +- .../upstream/rpcclient/JsonRpcParser.kt | 111 --------- .../upstream/rpcclient/JsonRpcRequest.kt | 53 ++-- .../upstream/rpcclient/JsonRpcResponse.kt | 11 + .../upstream/rpcclient/JsonRpcWsClient.kt | 30 +++ .../upstream/rpcclient/ResponseParser.kt | 182 ++++++++++++++ .../upstream/rpcclient/ResponseRpcParser.kt | 36 +++ .../upstream/rpcclient/ResponseWSParser.kt | 110 ++++++++ .../dshackle/test/EthereumApiMock.groovy | 234 +++++++++++++++++- .../ethereum/EthereumWsFactorySpec.groovy | 93 ++++++- ...ec.groovy => ResponseRpcParserSpec.groovy} | 27 +- .../rpcclient/ResponseWSParserSpec.groovy | 106 ++++++++ 18 files changed, 1003 insertions(+), 232 deletions(-) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParser.kt create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt rename src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/{JsonRpcParserSpec.groovy => ResponseRpcParserSpec.groovy} (94%) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParserSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt index dfdc9aa6..db3c8da7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt @@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspentDeserializer import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspent import io.emeraldpay.dshackle.upstream.bitcoin.data.RpcUnspentDeserializer +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import java.text.SimpleDateFormat import java.util.* @@ -48,6 +49,7 @@ class Global { module.addDeserializer(EsploraUnspent::class.java, EsploraUnspentDeserializer()) module.addDeserializer(RpcUnspent::class.java, RpcUnspentDeserializer()) + module.addDeserializer(JsonRpcRequest::class.java, JsonRpcRequest.Deserializer()) val objectMapper = ObjectMapper() objectMapper.registerModule(module) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index 1e0a9600..1844fa49 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -79,7 +79,7 @@ open class EthereumRpcUpstream( open fun createHead(): Head { return if (ethereumWsFactory != null) { - val ws = ethereumWsFactory.create(this).apply { + val ws = ethereumWsFactory.create().apply { connect() } val wsHead = EthereumWsHead(ws).apply { 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 e8e44c27..37977e31 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -23,25 +23,33 @@ import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson -import io.emeraldpay.etherjar.rpc.ws.SubscriptionJson +import io.netty.buffer.ByteBuf import io.netty.buffer.ByteBufInputStream +import io.netty.buffer.Unpooled import io.netty.handler.codec.http.HttpHeaderNames +import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono import reactor.core.publisher.Sinks +import reactor.core.scheduler.Schedulers import reactor.netty.http.client.HttpClient import reactor.netty.http.client.WebsocketClientSpec +import reactor.netty.http.websocket.WebsocketInbound +import reactor.netty.http.websocket.WebsocketOutbound import reactor.retry.Repeat -import java.io.InputStream +import reactor.util.function.Tuples import java.net.URI import java.time.Duration import java.util.* +import java.util.concurrent.Executors import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicReference +import java.util.concurrent.atomic.AtomicInteger + class EthereumWsFactory( private val uri: URI, @@ -50,27 +58,39 @@ class EthereumWsFactory( var basicAuth: AuthConfig.ClientBasicAuth? = null - fun create(upstream: EthereumUpstream): EthereumWs { - return EthereumWs(uri, origin, upstream, basicAuth) + fun create(): EthereumWs { + return EthereumWs(uri, origin, basicAuth) } class EthereumWs( private val uri: URI, private val origin: URI, - private val upstream: EthereumUpstream, private val basicAuth: AuthConfig.ClientBasicAuth? ) : AutoCloseable { companion object { private val log = LoggerFactory.getLogger(EthereumWs::class.java) + private const val IDS_START = 100 private const val START_REQUEST = "{\"jsonrpc\":\"2.0\", \"method\":\"eth_subscribe\", \"id\":\"blocks\", \"params\":[\"newHeads\"]}" } - private val topic = Sinks + private val parser = ResponseWSParser() + + private val blocks = Sinks .many() .multicast() .directBestEffort() + private val rpcSend = Sinks + .many() + .unicast() + .onBackpressureBuffer() + private val rpcReceive = Sinks + .many() + .multicast() + .directBestEffort() + private val sendIdSeq = AtomicInteger(IDS_START) + private val sendExecutor = Executors.newSingleThreadExecutor() private var keepConnection = true private var connection: Disposable? = null @@ -89,11 +109,6 @@ class EthereumWsFactory( private fun connectInternal() { log.info("Connecting to WebSocket: $uri") connection?.dispose() - connection = null - - val subscriptionId = AtomicReference("NOTSET") - - val objectMapper = Global.objectMapper connection = HttpClient.create() .doOnError( { _, t -> @@ -101,9 +116,7 @@ class EthereumWsFactory( // going to try to reconnect later tryReconnectLater() }, - { _, _ -> - - } + { _, _ -> } ) .headers { headers -> headers.add(HttpHeaderNames.ORIGIN, origin) @@ -114,11 +127,7 @@ class EthereumWsFactory( } } .let { - if (uri.scheme == "wss") { - it.secure() - } else { - it - } + if (uri.scheme == "wss") it.secure() else it } .websocket( WebsocketClientSpec.builder() @@ -128,57 +137,93 @@ class EthereumWsFactory( ) .uri(uri) .handle { inbound, outbound -> - val consumer = inbound.aggregateFrames() - .aggregateFrames(8 * 65_536) - .receiveFrames() - .flatMap { - val msg: SubscriptionJson = objectMapper.readerFor(SubscriptionJson::class.java) - .readValue(ByteBufInputStream(it.content()) as InputStream) - when { - msg.error != null -> { - Mono.error(IllegalStateException("Received error from WS upstream")) - } - msg.subscription == subscriptionId.get() -> { - onNewBlock(msg.blockResult) - Mono.empty() - } - msg.subscription == null -> { - // received ID for subscription - subscriptionId.set(msg.result.asText()) - log.debug("Connected to $uri") - Mono.empty() - } - else -> { - Mono.error(IllegalStateException("Unknown message received: ${msg.subscription}")) - } - } - } - .onErrorResume { t -> - log.warn("Connection dropped to $uri. Error: ${t.message}") - // going to try to reconnect later - tryReconnectLater() - // completes current outbound flow - Mono.empty() - } - - - outbound.sendString(Mono.just(START_REQUEST) - .doOnError { log.warn("Failed to start WS subscription. ${it.javaClass}: ${it.message}") }) - .then(consumer.then()) + handle(inbound, outbound) } .doOnError { - println(it) + log.error("Failed to setup WS connection", it) } .subscribe() } - fun onNewBlock(block: BlockJson) { - // WS returns incomplete blocks, i.e. without some fields, so need to fetch full block data - if (block.difficulty == null || block.transactions == null) { + fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher { + val consumer = inbound.aggregateFrames() + // accept up to 1Mb messages + .aggregateFrames(16 * 65_536) + .receiveFrames() + .map { ByteBufInputStream(it.content()).readAllBytes() } + .flatMap { + try { + val msg = parser.parse(it) + if (msg.type == ResponseWSParser.Type.SUBSCRIPTION) { + onSubscription(msg) + } else { + onRpc(msg) + } + } catch (t: Throwable) { + log.warn("Failed to process WS message. ${t.javaClass}: ${t.message}") + Mono.empty() + } + } + .onErrorResume { t -> + log.warn("Connection dropped to $uri. Error: ${t.message}") + // going to try to reconnect later + tryReconnectLater() + // completes current outbound flow + Mono.empty() + } + + val start = Mono.just(START_REQUEST).map { + Unpooled.wrappedBuffer(it.toByteArray()) + } + val calls = rpcSend + .asFlux() + .map { + Unpooled.wrappedBuffer(Global.objectMapper.writeValueAsBytes(it)) + } + + return outbound.send( + Flux.merge( + start, + calls.subscribeOn(Schedulers.boundedElastic()), + consumer.then(Mono.empty()).subscribeOn(Schedulers.boundedElastic()) + ) + ) + } + + fun onRpc(msg: ResponseWSParser.WsResponse): Mono { + return if (msg.id.isNumber()) { + val resp = JsonRpcResponse( + msg.value, msg.error, msg.id + ) + Mono.fromCallable { + val status = rpcReceive.tryEmitNext(resp) + if (status.isFailure) { + log.warn("Failed to proceed with a RPC message: $status") + } + }.then() + } else { + //it's a response to the newHeads subscription, just ignore it + Mono.empty() + } + } + + fun onSubscription(msg: ResponseWSParser.WsResponse): Mono { + if (msg.error != null) { + return Mono.error(IllegalStateException("Received error from WS upstream: ${msg.error.message}")) + } + // we always expect an answer to the `newHeads`, since we are not initiating any other subscriptions + return Mono.fromCallable { + Global.objectMapper.readValue(msg.value, BlockJson::class.java) as BlockJson + }.flatMap { onNewHeads(it) }.then() + } + + fun onNewHeads(block: BlockJson): Mono { + // newHeads returns incomplete blocks, i.e. without some fields and without transaction hashes, + // so we need to fetch the full block data + return if (block.difficulty == null || block.transactions == null) { Mono.just(block.hash) .flatMap { hash -> - upstream.getApi() - .read(JsonRpcRequest("eth_getBlockByHash", listOf(hash.toHex(), false))) + call(JsonRpcRequest("eth_getBlockByHash", listOf(hash.toHex(), false))) .flatMap { resp -> if (resp.isNull()) { Mono.error(SilentException("Received null for block $hash")) @@ -188,6 +233,8 @@ class EthereumWsFactory( } .flatMap(JsonRpcResponse::requireResult) .map { BlockContainer.fromEthereumJson(it) } + .subscribeOn(Schedulers.boundedElastic()) + .timeout(Defaults.timeoutInternal, Mono.empty()) }.repeatWhenEmpty { n -> Repeat.times(5) .exponentialBackoff(Duration.ofMillis(50), Duration.ofMillis(500)) @@ -195,17 +242,53 @@ class EthereumWsFactory( } .timeout(Defaults.timeout, Mono.empty()) .onErrorResume { Mono.empty() } - .subscribe { - topic.tryEmitNext(it) + .doOnNext { + blocks.tryEmitNext(it) } - + .then() } else { - topic.tryEmitNext(BlockContainer.from(block)) + Mono.fromCallable { + blocks.tryEmitNext(BlockContainer.from(block)) + }.then() } } - fun getFlux(): Flux { - return this.topic.asFlux() + fun call(originalRequest: JsonRpcRequest): Mono { + return Mono.fromCallable { + // use an internal id sequence, to avoid id conflicts with user calls + val internalId = sendIdSeq.getAndIncrement() + val originalId = originalRequest.id + Tuples.of(originalRequest.copy(id = internalId), originalId) + }.flatMap { request -> + waitForResponse(request.t1, request.t2) + } + } + + fun sendRpc(request: JsonRpcRequest) { + // submit to upstream in a separate thread, to free current thread (needs for subscription, etc) + sendExecutor.execute { + val result = rpcSend.tryEmitNext(request) + if (result.isFailure) { + log.warn("Failed to send RPC request: $result") + } + } + } + + fun waitForResponse(request: JsonRpcRequest, originalId: Int): Mono { + val expectedId = request.id.toLong() + return Mono.just(request) + .flatMap { + Flux.from(rpcReceive.asFlux()) + .doOnSubscribe { sendRpc(request) } + .filter { resp -> resp.id.asNumber() == expectedId } + .take(1) + .singleOrEmpty() + .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) } + } + } + + fun getBlocksFlux(): Flux { + return this.blocks.asFlux() } override fun close() { @@ -216,5 +299,4 @@ class EthereumWsFactory( } - } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index 4abe9284..3386611e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -34,7 +34,7 @@ class EthereumWsHead( override fun start() { this.subscription?.dispose() - this.subscription = super.follow(ws.getFlux()) + this.subscription = super.follow(ws.getBlocksFlux()) } override fun stop() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt index cea25456..e05813ec 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt @@ -123,8 +123,8 @@ class NativeCallRouter( } } - fun getBlockByNumber(params: List): Mono? { - if (params.size != 2) { + fun getBlockByNumber(params: List): Mono? { + if (params.size != 2 || params[0] == null || params[1] == null) { throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters") } val number: Long @@ -155,7 +155,7 @@ class NativeCallRouter( } } } catch (e: IllegalArgumentException) { - throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block number") + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be a block number") } val withTx = params[1].toString().toBoolean() var block = reader.blocksByHeightAsCont() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt index 66791a88..50baa980 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt @@ -52,5 +52,9 @@ class JsonRpcError(val code: Int, val message: String, val details: Any?) { return result } + override fun toString(): String { + return "JsonRpcError(code=$code, message='$message', details=$details)" + } + } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt index ce50a5a6..5fbf6b04 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt @@ -19,8 +19,6 @@ import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.etherjar.rpc.RpcException import io.emeraldpay.etherjar.rpc.RpcResponseError -import io.micrometer.core.instrument.Counter -import io.micrometer.core.instrument.Timer import io.netty.buffer.Unpooled import io.netty.handler.codec.http.HttpHeaderNames import io.netty.handler.codec.http.HttpHeaders @@ -50,7 +48,7 @@ class JsonRpcHttpClient( private val log = LoggerFactory.getLogger(JsonRpcHttpClient::class.java) } - private val parser = JsonRpcParser() + private val parser = ResponseRpcParser() private val httpClient: HttpClient init { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt deleted file mode 100644 index 17fbad91..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParser.kt +++ /dev/null @@ -1,111 +0,0 @@ -/** - * Copyright (c) 2020 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.rpcclient - -import com.fasterxml.jackson.core.JsonFactory -import com.fasterxml.jackson.core.JsonParseException -import com.fasterxml.jackson.core.JsonParser -import com.fasterxml.jackson.core.JsonToken -import io.emeraldpay.dshackle.Global -import io.emeraldpay.etherjar.rpc.RpcResponseError -import org.slf4j.LoggerFactory - -class JsonRpcParser() { - - companion object { - private val log = LoggerFactory.getLogger(JsonRpcParser::class.java) - } - - private val jsonFactory = JsonFactory() - - fun parse(json: ByteArray): JsonRpcResponse { - try { - val parser: JsonParser = jsonFactory.createParser(json) - parser.nextToken() - if (parser.currentToken != JsonToken.START_OBJECT) { - return JsonRpcResponse(null, JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON")) - } - var nullResponse: JsonRpcResponse? = null - while (parser.nextToken() != JsonToken.END_OBJECT) { - val field = parser.currentName - if (field == "jsonrpc" || field == "id") { - if (!parser.nextToken().isScalarValue) { - return JsonRpcResponse(null, JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (id or jsonrpc value)")) - } - // just skip the field - } else if (field == "result") { - val value = parser.nextToken() - val start = parser.tokenLocation - if (value.isScalarValue) { - val text = parser.text - if (value == JsonToken.VALUE_STRING) { - return JsonRpcResponse(("\"" + text + "\"").toByteArray(), null) - } else if (value == JsonToken.VALUE_NULL) { - //if null we should check if error is present - nullResponse = JsonRpcResponse(text.toByteArray(), null) - } else { - return JsonRpcResponse(text.toByteArray(), null) - } - } else if (value == JsonToken.START_OBJECT || value == JsonToken.START_ARRAY) { - parser.skipChildren() - val end = parser.currentLocation.byteOffset.toInt() - val copy = ByteArray((end - start.byteOffset).toInt()) - System.arraycopy(json, start.byteOffset.toInt(), copy, 0, copy.size) - return JsonRpcResponse(copy, null) - } - } else if (field == "error") { - val err = readError(parser) - if (err != null) { - return JsonRpcResponse(null, err) - } - } - } - if (nullResponse != null) { - return nullResponse - } - } catch (e: JsonParseException) { - log.warn("Failed to parse JSON from upstream: ${e.message}") - } - return JsonRpcResponse(null, JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON structure")) - } - - fun readError(parser: JsonParser): JsonRpcError? { - var code = 0 - var message = "" - var details: Any? = null - - while (parser.nextToken() != JsonToken.END_OBJECT) { - if (parser.currentToken() == JsonToken.VALUE_NULL) { - // error is just null - return null - } - val field = parser.currentName() - if (field == "code" && parser.currentToken == JsonToken.VALUE_NUMBER_INT) { - code = parser.intValue - } else if (field == "message" && parser.currentToken == JsonToken.VALUE_STRING) { - message = parser.valueAsString - } else if (field == "data") { - when (val value = parser.nextToken()) { - JsonToken.VALUE_NULL -> details = null - JsonToken.VALUE_STRING -> details = parser.valueAsString - JsonToken.START_OBJECT -> details = Global.objectMapper.readValue(parser, java.util.Map::class.java) - else -> log.warn("Unsupported error data type $value") - } - } - } - return JsonRpcError(code, message, details) - } -} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt index 5339468e..bef4df17 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcRequest.kt @@ -15,40 +15,55 @@ */ package io.emeraldpay.dshackle.upstream.rpcclient +import com.fasterxml.jackson.core.JsonParser +import com.fasterxml.jackson.databind.DeserializationContext +import com.fasterxml.jackson.databind.JsonDeserializer +import com.fasterxml.jackson.databind.JsonNode import io.emeraldpay.dshackle.Global -class JsonRpcRequest( +data class JsonRpcRequest( val method: String, - val params: List + val params: List, + val id: Int ) { + constructor(method: String, params: List) : this(method, params, 1) + fun toJson(): ByteArray { val json = mapOf( "jsonrpc" to "2.0", - "id" to 1, + "id" to id, "method" to method, "params" to params ) return Global.objectMapper.writeValueAsBytes(json) } - override fun equals(other: Any?): Boolean { - if (this === other) return true - if (other !is JsonRpcRequest) return false - - if (method != other.method) return false - if (params != other.params) return false - - return true - } - - override fun hashCode(): Int { - var result = method.hashCode() - result = 31 * result + params.hashCode() - return result - } - override fun toString(): String { return String(this.toJson()) } + + class Deserializer : JsonDeserializer() { + + override fun deserialize(p: JsonParser, ctxt: DeserializationContext): JsonRpcRequest { + val node: JsonNode = p.readValueAsTree() + val id = node.get("id").intValue() + val method = node.get("method").textValue() + val params = node.get("params").map { + if (it.isNumber) { + it.asInt() + } else if (it.isTextual) { + it.textValue() + } else if (it.isBoolean) { + it.booleanValue() + } else if (it.isNull) { + null + } else { + throw IllegalStateException("Unsupported param type: ${it.asToken()}") + } + } + return JsonRpcRequest(method, params, id) + } + + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt index 03ed29e8..63fb1e67 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt @@ -106,6 +106,10 @@ class JsonRpcResponse( } } + fun copyWithId(id: Id): JsonRpcResponse { + return JsonRpcResponse(result, error, id) + } + override fun equals(other: Any?): Boolean { if (this === other) return true if (other !is JsonRpcResponse) return false @@ -177,6 +181,10 @@ class JsonRpcResponse( override fun hashCode(): Int { return id.hashCode() } + + override fun toString(): String { + return id.toString() + } } class StringId(val id: String) : Id { @@ -205,6 +213,9 @@ class JsonRpcResponse( return id.hashCode() } + override fun toString(): String { + return id + } } class ResponseJsonSerializer : JsonSerializer() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt new file mode 100644 index 00000000..9bed273d --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt @@ -0,0 +1,30 @@ +/** + * 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.upstream.rpcclient + +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory +import reactor.core.publisher.Mono + +class JsonRpcWsClient( + private val ws: EthereumWsFactory.EthereumWs +) : Reader { + + override fun read(key: JsonRpcRequest): Mono { + return ws.call(key) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt new file mode 100644 index 00000000..af75484b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt @@ -0,0 +1,182 @@ +/** + * 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.upstream.rpcclient + +import com.fasterxml.jackson.core.JsonFactory +import com.fasterxml.jackson.core.JsonParseException +import com.fasterxml.jackson.core.JsonParser +import com.fasterxml.jackson.core.JsonToken +import io.emeraldpay.dshackle.Global +import io.emeraldpay.etherjar.rpc.RpcResponseError +import org.slf4j.LoggerFactory +import java.io.IOException + +abstract class ResponseParser { + + companion object { + private val log = LoggerFactory.getLogger(ResponseParser::class.java) + } + + private val jsonFactory = JsonFactory() + + abstract fun build(state: Preparsed): T + + fun parse(json: ByteArray): T { + return build(parseInternal(json)) + } + + private fun parseInternal(json: ByteArray): Preparsed { + var state = Preparsed() + try { + val parser: JsonParser = jsonFactory.createParser(json) + parser.nextToken() + if (parser.currentToken != JsonToken.START_OBJECT) { + return Preparsed(error = JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON: not an Object")) + } + while (parser.nextToken() != JsonToken.END_OBJECT) { + val field = parser.currentName + state = process(parser, json, field, state) + } + } catch (e: JsonParseException) { + log.warn("Failed to parse JSON from upstream: ${e.message}") + } + if (state.isReady) { + return state + } + return Preparsed(error = JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON structure: never finalized")) + } + + open fun process(parser: JsonParser, json: ByteArray, field: String, state: Preparsed): Preparsed { + if (field == "jsonrpc") { + if (!parser.nextToken().isScalarValue) { + return state.copy(error = JsonRpcError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (jsonrpc value)")) + } + // just skip the field + return state + } else if (field == "id") { + return state.copy(id = readId(parser)) + } else if (field == "result") { + val result = readResult(json, parser) + return if (result == null) { + //if result is null we should check if an error is also present, and if it's set then return only the error + state.copy(nullResult = true) + } else { + state.copy(result = result) + } + } else if (field == "error") { + val err = readError(parser) + if (err != null) { + return state.copy(error = err) + } + } + return state + } + + private fun readId(parser: JsonParser): JsonRpcResponse.Id { + if (parser.currentToken() == JsonToken.FIELD_NAME) { + parser.nextToken() + } + return if (parser.currentToken() == JsonToken.VALUE_NUMBER_INT) { + JsonRpcResponse.NumberId(parser.intValue) + } else if (parser.currentToken() == JsonToken.VALUE_STRING) { + JsonRpcResponse.StringId(parser.text) + } else { + throw IllegalStateException("Not a string or number: ${parser.currentToken()}") + } + } + + @Throws(IOException::class) + private fun readNumber(parser: JsonParser): Int { + if (parser.currentToken() != JsonToken.VALUE_NUMBER_INT) { + parser.nextToken() + } + if (!parser.currentToken().isNumeric) { + throw IllegalStateException("Not a number: ${parser.currentToken.name}") + } + return parser.intValue + } + + fun readResult(json: ByteArray, parser: JsonParser): ByteArray? { + val value = parser.nextToken() + val start = parser.tokenLocation + if (value.isScalarValue) { + val text = parser.text + return if (value == JsonToken.VALUE_STRING) { + ("\"" + text + "\"").toByteArray() + } else if (value == JsonToken.VALUE_NULL) { + null + } else { + text.toByteArray() + } + } else if (value == JsonToken.START_OBJECT || value == JsonToken.START_ARRAY) { + parser.skipChildren() + val end = parser.currentLocation.byteOffset.toInt() + val copy = ByteArray((end - start.byteOffset).toInt()) + System.arraycopy(json, start.byteOffset.toInt(), copy, 0, copy.size) + return copy + } else { + throw IllegalStateException("Invalid JSON structure, cannot read result from ${value.name}") + } + } + + fun readError(parser: JsonParser): JsonRpcError? { + var code = 0 + var message = "" + var details: Any? = null + + while (parser.nextToken() != JsonToken.END_OBJECT) { + if (parser.currentToken() == JsonToken.VALUE_NULL) { + // error is just null + return null + } + val field = parser.currentName() + if (field == "code" && parser.currentToken == JsonToken.VALUE_NUMBER_INT) { + code = parser.intValue + } else if (field == "message" && parser.currentToken == JsonToken.VALUE_STRING) { + message = parser.valueAsString + } else if (field == "data") { + when (val value = parser.nextToken()) { + JsonToken.VALUE_NULL -> details = null + JsonToken.VALUE_STRING -> details = parser.valueAsString + JsonToken.START_OBJECT -> details = Global.objectMapper.readValue(parser, java.util.Map::class.java) + else -> log.warn("Unsupported error data type $value") + } + } + } + return JsonRpcError(code, message, details) + } + + data class Preparsed( + val id: JsonRpcResponse.Id? = null, + val result: ByteArray? = null, + val nullResult: Boolean = false, + val error: JsonRpcError? = null, + val subMethod: String? = null, + val subId: String? = null + ) { + + private val isResultSet = result != null || nullResult + + val isRpcReady: Boolean = id != null && + (error != null || isResultSet) + + val isSubReady: Boolean = subId != null && + isResultSet + + val isReady: Boolean = isRpcReady || isSubReady + + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParser.kt new file mode 100644 index 00000000..7db80ace --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParser.kt @@ -0,0 +1,36 @@ +/** + * Copyright (c) 2020 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.rpcclient + +import org.slf4j.LoggerFactory + +open class ResponseRpcParser() : ResponseParser() { + + companion object { + private val log = LoggerFactory.getLogger(ResponseRpcParser::class.java) + } + + override fun build(state: Preparsed): JsonRpcResponse { + if (state.error != null) { + return JsonRpcResponse(null, state.error, state.id ?: JsonRpcResponse.Id.from(-1)) + } + if (state.nullResult) { + return JsonRpcResponse("null".toByteArray(), null, state.id ?: JsonRpcResponse.Id.from(-1)) + } + return JsonRpcResponse(state.result, null, state.id ?: JsonRpcResponse.Id.from(-1)) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt new file mode 100644 index 00000000..07e7e654 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt @@ -0,0 +1,110 @@ +/** + * 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.upstream.rpcclient + +import com.fasterxml.jackson.core.JsonParser +import com.fasterxml.jackson.core.JsonToken +import org.slf4j.LoggerFactory +import java.io.IOException + + +class ResponseWSParser : ResponseParser() { + + companion object { + private val log = LoggerFactory.getLogger(ResponseWSParser::class.java) + private val NULL_RESULT = "null".toByteArray() + } + + override fun build(state: Preparsed): WsResponse { + if (state.isRpcReady) { + return WsResponse( + Type.RPC, + state.id!!, + if (state.nullResult) NULL_RESULT else state.result, + state.error + ) + } + if (state.isSubReady) { + return WsResponse( + Type.SUBSCRIPTION, + JsonRpcResponse.Id.from(state.subId!!), + if (state.nullResult) NULL_RESULT else state.result, + state.error + ) + } + throw IllegalStateException("State is not ready") + } + + override fun process(parser: JsonParser, json: ByteArray, field: String, state: Preparsed): Preparsed { + if ("method" == field) { + parser.nextToken() + val method = parser.getValueAsString() + return state.copy(subMethod = method) + } + if ("params" == field) { + // example: + // newHeads + // { + // "jsonrpc": "2.0", + // "method": "eth_subscription", + // "params": { + // "result": { + // "difficulty": ...... + // }, + // "subscription": "...." + // } + //} + return decodeSubscription(parser, json, state) + } + return super.process(parser, json, field, state) + } + + @Throws(IOException::class) + private fun decodeString(parser: JsonParser): String { + if (parser.currentToken() != JsonToken.VALUE_STRING) { + parser.nextToken() + } + check(parser.currentToken().isScalarValue) { "Id is not a string" } + return parser.valueAsString + } + + @Throws(IOException::class) + protected fun decodeSubscription(parser: JsonParser, json: ByteArray, stateOriginal: Preparsed): Preparsed { + var state = stateOriginal + while (parser.nextToken() != JsonToken.END_OBJECT) { + checkNotNull(parser.currentToken()) { "JSON finished before data received" } + val field = parser.currentName() + if ("subscription" == field) { + state = state.copy(subId = decodeString(parser)) + } else if ("result" == field) { + state = state.copy(result = readResult(json, parser)) + } + } + return state + } + + enum class Type { + SUBSCRIPTION, RPC + } + + data class WsResponse( + val type: Type, + val id: JsonRpcResponse.Id, + val value: ByteArray?, + val error: JsonRpcError? + ) + +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy index b92e3ec7..eb4224ee 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumApiMock.groovy @@ -27,13 +27,34 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.grpc.stub.StreamObserver import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.etherjar.rpc.json.ResponseJson +import io.netty.buffer.ByteBuf +import io.netty.buffer.ByteBufAllocator +import io.netty.buffer.ByteBufInputStream +import io.netty.buffer.Unpooled +import io.netty.handler.codec.http.HttpHeaders +import io.netty.handler.codec.http.websocketx.TextWebSocketFrame +import io.netty.handler.codec.http.websocketx.WebSocketCloseStatus +import io.netty.handler.codec.http.websocketx.WebSocketFrame import org.jetbrains.annotations.NotNull +import org.reactivestreams.Publisher import org.slf4j.Logger import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import reactor.core.publisher.Sinks +import reactor.netty.ByteBufFlux +import reactor.netty.Connection +import reactor.netty.NettyInbound +import reactor.netty.NettyOutbound +import reactor.netty.http.websocket.WebsocketInbound +import reactor.netty.http.websocket.WebsocketOutbound +import reactor.util.annotation.Nullable import java.time.Duration import java.util.concurrent.Callable +import java.util.function.BiFunction +import java.util.function.Consumer +import java.util.function.Predicate class EthereumApiMock implements Reader { @@ -57,7 +78,7 @@ class EthereumApiMock implements Reader { } @Override - Mono read(JsonRpcRequest request) { + Mono read(JsonRpcRequest request, boolean required = true) { Callable call = { def predefined = predefined.find { it.isSame(request.method, request.params) } byte[] result = null @@ -65,7 +86,7 @@ class EthereumApiMock implements Reader { if (predefined != null) { if (predefined.exception != null) { predefined.onCalled() - predefined.print() + predefined.print(request.id) throw predefined.exception } if (predefined.result instanceof RpcResponseError) { @@ -77,12 +98,15 @@ class EthereumApiMock implements Reader { result = objectMapper.writeValueAsBytes(predefined.result) } predefined.onCalled() - predefined.print() + predefined.print(request.id) } else { log.error("Method ${request.method} with ${request.params} is not mocked") + if (!required) { + return null + } error = new JsonRpcError(-32601, "Method ${request.method} with ${request.params} is not mocked") } - return new JsonRpcResponse(result, error) + return new JsonRpcResponse(result, error, JsonRpcResponse.Id.from(request.id)) } as Callable return Mono.fromCallable(call) } @@ -104,6 +128,10 @@ class EthereumApiMock implements Reader { responseObserver.onCompleted() } + WebsocketApi asWebsocket() { + return new WebsocketApi(this) + } + class PredefinedResponse { String method List params @@ -132,8 +160,202 @@ class EthereumApiMock implements Reader { } } - void print() { - println "Execute API: $method ${params ? params : '_'} >> $result" + void print(int id) { + println "Execute API: $id $method ${params ? params : '_'} >> $result" + } + } + + class WebsocketApi { + private final EthereumApiMock api + + private Sinks.Many responses = Sinks + .many() + .unicast() + .onBackpressureBuffer() + private Sinks.Many jsonResponses = Sinks + .many() + .unicast() + .onBackpressureBuffer() + private WebsocketOutboundMock outbound + private WebsocketInboundMock inbound + + WebsocketApi(EthereumApiMock api) { + this.api = api + outbound = new WebsocketOutboundMock(api, responses) + inbound = new WebsocketInboundMock(responses.asFlux(), jsonResponses.asFlux()) + } + + boolean send(String json) { + jsonResponses.tryEmitNext(json).success + } + + WebsocketOutbound getOutbound() { + return outbound + } + + WebsocketInbound getInbound() { + return inbound + } + } + + class WebsocketInboundMock implements WebsocketInbound { + + private final Flux responses + private final Flux jsonResponses + + WebsocketInboundMock(Flux responses, Flux jsonResponses) { + this.responses = responses + this.jsonResponses = jsonResponses + } + + @Override + String selectedSubprotocol() { + throw new UnsupportedOperationException() + } + + @Override + HttpHeaders headers() { + throw new UnsupportedOperationException() + } + + @Override + Mono receiveCloseStatus() { + return Mono.empty() + } + + @Override + ByteBufFlux receive() { + throw new UnsupportedOperationException() + } + + @Override + Flux receiveObject() { + throw new UnsupportedOperationException() + } + + @Override + NettyInbound withConnection(Consumer withConnection) { + return this + } + + @Override + Flux receiveFrames() { + return Flux.merge( + jsonResponses, + responses.map { + Global.objectMapper.writeValueAsString(it) + }) + .map { + println("WS server->client msg: $it") + new TextWebSocketFrame(it) + } + .doOnError { t -> + t.printStackTrace() + } + } + } + + class WebsocketOutboundMock implements WebsocketOutbound { + + private final EthereumApiMock api + private final Sinks.Many responses + + WebsocketOutboundMock(EthereumApiMock api, Sinks.Many responses) { + this.api = api + this.responses = responses + } + + @Override + String selectedSubprotocol() { + throw new UnsupportedOperationException() + } + + @Override + ByteBufAllocator alloc() { + throw new UnsupportedOperationException() + } + + private void handle(Publisher dataStream) { + Flux.from(dataStream) + .map { it -> + Global.objectMapper.readValue(new ByteBufInputStream(it), JsonRpcRequest) + } + .flatMap { JsonRpcRequest request -> + api.read(request, false) + } + .doOnNext { + def status = responses.tryEmitNext(it) + if (status.isFailure()) { + println("Failed to send through mock: $status") + } + } + .subscribe() + } + + @Override + NettyOutbound send(Publisher dataStream) { + handle(dataStream) + return this + } + + @Override + NettyOutbound send(Publisher dataStream, Predicate predicate) { + handle(dataStream) + return this + } + + @Override + NettyOutbound sendObject(Publisher dataStream, Predicate predicate) { + def msgs = Flux.from(dataStream) + .cast(TextWebSocketFrame) + .map { + Unpooled.wrappedBuffer(it.text().bytes) + } + handle(msgs) + return this + } + + @Override + NettyOutbound sendObject(Object message) { + return this + } + + @Override + def NettyOutbound sendUsing(Callable sourceInput, BiFunction mappedInput, Consumer sourceCleanup) { + return this + } + + @Override + NettyOutbound withConnection(Consumer withConnection) { + return this + } + + @Override + Mono sendClose() { + return Mono.fromCallable { + responses.tryEmitComplete() + }.then() + } + + @Override + Mono sendClose(int rsv) { + return Mono.fromCallable { + responses.tryEmitComplete() + }.then() + } + + @Override + Mono sendClose(int statusCode, @Nullable String reasonText) { + return Mono.fromCallable { + responses.tryEmitComplete() + }.then() + } + + @Override + Mono sendClose(int rsv, int statusCode, @Nullable String reasonText) { + return Mono.fromCallable { + responses.tryEmitComplete() + }.then() } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy index 32e6615d..f5415999 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy @@ -15,11 +15,15 @@ */ package io.emeraldpay.dshackle.upstream.ethereum -import io.emeraldpay.dshackle.cache.BlocksMemCache +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.etherjar.domain.TransactionId +import io.emeraldpay.etherjar.rpc.RpcResponseError import io.emeraldpay.etherjar.rpc.json.BlockJson +import io.emeraldpay.etherjar.rpc.json.TransactionJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import reactor.core.publisher.Flux import reactor.test.StepVerifier @@ -34,7 +38,6 @@ class EthereumWsFactorySpec extends Specification { def "Fetch block"() { setup: def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) - def blocksCache = Mock(BlocksMemCache) def block = new BlockJson() block.number = 100 @@ -44,20 +47,98 @@ class EthereumWsFactorySpec extends Specification { block.uncles = [] block.totalDifficulty = BigInteger.ONE + def headBlock = block.copy().tap { + it.transactions = null + } + def apiMock = TestingCommons.api() - def upstream = TestingCommons.upstream(apiMock) - def ws = wsf.create(upstream) + def wsApiMock = apiMock.asWebsocket() + def ws = wsf.create() apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block) when: - def act = Flux.from(ws.getFlux()) + Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() + def act = Flux.from(ws.getBlocksFlux()) then: StepVerifier.create(act) - .then { ws.onNewBlock(block) } + .then { ws.onNewHeads(headBlock).subscribe() } .expectNext(BlockContainer.from(block)) .thenCancel() .verify(Duration.ofSeconds(1)) } + + def "Makes a RPC call"() { + setup: + def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def apiMock = TestingCommons.api() + def wsApiMock = apiMock.asWebsocket() + def ws = wsf.create() + + def tx = new TransactionJson().tap { + hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") + } + apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], tx) + + when: + Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() + def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15)) + + then: + StepVerifier.create(act) + .expectNextMatches { + it.id.asNumber() == 15L && Global.objectMapper.readValue(it.result, TransactionJson) == tx + } + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Makes a RPC call - return null"() { + setup: + def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def apiMock = TestingCommons.api() + def wsApiMock = apiMock.asWebsocket() + def ws = wsf.create() + + apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null) + + when: + Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() + def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15)) + + then: + StepVerifier.create(act) + .expectNextMatches { + it.id.asNumber() == 15L && + it.resultAsRawString == 'null' + } + .expectComplete() + .verify(Duration.ofSeconds(1)) + } + + def "Makes a RPC call - return error"() { + setup: + def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) + def apiMock = TestingCommons.api() + def wsApiMock = apiMock.asWebsocket() + def ws = wsf.create() + + apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], + new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test")) + + when: + Flux.from(ws.handle(wsApiMock.inbound, wsApiMock.outbound)).subscribe() + def act = ws.call(new JsonRpcRequest("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], 15)) + + then: + StepVerifier.create(act) + .expectNextMatches { + it.id.asNumber() == 15L && + it.error != null && + it.error.code == RpcResponseError.CODE_METHOD_NOT_EXIST && it.error.message == "test" + } + .expectComplete() + .verify(Duration.ofSeconds(1)) + } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParserSpec.groovy similarity index 94% rename from src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy rename to src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParserSpec.groovy index c07f8444..f2f3542d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcParserSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseRpcParserSpec.groovy @@ -18,19 +18,9 @@ package io.emeraldpay.dshackle.upstream.rpcclient import io.emeraldpay.etherjar.rpc.RpcResponseError import spock.lang.Specification -class JsonRpcParserSpec extends Specification { +class ResponseRpcParserSpec extends Specification { - JsonRpcParser parser = new JsonRpcParser() - - def "Parse just result"() { - setup: - def json = '{"result": "Hello world!"}' - when: - def act = parser.parse(json.getBytes()) - then: - act.error == null - new String(act.result) == '"Hello world!"' - } + ResponseRpcParser parser = new ResponseRpcParser() def "Parse string response"() { setup: @@ -178,6 +168,19 @@ class JsonRpcParserSpec extends Specification { !act.hasResult() } + def "Parse error with no result field"() { + setup: + def json = '{"jsonrpc": "2.0", "id": 1, "error": {"code": -1111, "message": "test"}}' + when: + def act = parser.parse(json.getBytes()) + then: + act.error != null + act.error.code == -1111 + act.error.message == "test" + act.hasError() + !act.hasResult() + } + def "Parse error with data"() { setup: // 0 8 16 32 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParserSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParserSpec.groovy new file mode 100644 index 00000000..460e5e02 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParserSpec.groovy @@ -0,0 +1,106 @@ +/** + * 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.upstream.rpcclient + +import spock.lang.Specification + +class ResponseWSParserSpec extends Specification { + + ResponseWSParser parser = new ResponseWSParser() + + def "Parse subscription response"() { + setup: + def msg = "{\n" + + " \"id\": \"blocks\", \n" + + " \"jsonrpc\": \"2.0\", \n" + + " \"result\": \"0x9cef478923ff08bf67fde6c64013158d\"\n" + + "}" + when: + def act = parser.parse(msg.bytes) + then: + act.type == ResponseWSParser.Type.RPC + act.id.asString() == "blocks" + act.error == null + act.value == "\"0x9cef478923ff08bf67fde6c64013158d\"".bytes + } + + def "Parse newHeads event"() { + setup: + def msg = "{\n" + + " \"jsonrpc\": \"2.0\",\n" + + " \"method\": \"eth_subscription\",\n" + + " \"params\": {\n" + + " \"result\": {\n" + + " \"difficulty\": \"0x15d9223a23aa\",\n" + + " \"extraData\": \"0xd983010305844765746887676f312e342e328777696e646f7773\",\n" + + " \"gasLimit\": \"0x47e7c4\",\n" + + " \"gasUsed\": \"0x38658\",\n" + + " \"logsBloom\": \"0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000\",\n" + + " \"miner\": \"0xf8b483dba2c3b7176a3da549ad41a48bb3121069\",\n" + + " \"nonce\": \"0x084149998194cc5f\",\n" + + " \"number\": \"0x1348c9\",\n" + + " \"parentHash\": \"0x7736fab79e05dc611604d22470dadad26f56fe494421b5b333de816ce1f25701\",\n" + + " \"receiptRoot\": \"0x2fab35823ad00c7bb388595cb46652fe7886e00660a01e867824d3dceb1c8d36\",\n" + + " \"sha3Uncles\": \"0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347\",\n" + + " \"stateRoot\": \"0xb3346685172db67de536d8765c43c31009d0eb3bd9c501c9be3229203f15f378\",\n" + + " \"timestamp\": \"0x56ffeff8\",\n" + + " \"transactionsRoot\": \"0x0167ffa60e3ebc0b080cdb95f7c0087dd6c0e61413140e39d94d3468d7c9689f\"\n" + + " },\n" + + " \"subscription\": \"0x9ce59a13059e417087c02d3236a0b1cc\"\n" + + " }\n" + + "}" + when: + def act = parser.parse(msg.bytes) + then: + act.type == ResponseWSParser.Type.SUBSCRIPTION + act.id.asString() == "0x9ce59a13059e417087c02d3236a0b1cc" + act.error == null + with(new String(act.value)) { + it.length() > 0 + it.startsWith("{") + it.endsWith("}") + it.contains("\"difficulty\": \"0x15d9223a23aa\"") + } + } + + def "Parse RPC with error"() { + setup: + def msg = "{\"jsonrpc\":\"2.0\",\"id\":151,\"error\":{\"code\":-32602,\"message\":\"invalid blocknumber\"}}" + when: + def act = parser.parse(msg.bytes) + then: + act.type == ResponseWSParser.Type.RPC + act.id.asNumber() == 151L + act.error != null + act.value == null + with(act.error) { + it.code == -32602 + it.message == "invalid blocknumber" + } + } + + def "Parse RPC with null result"() { + setup: + def msg = "{\"jsonrpc\":\"2.0\",\"id\":100,\"result\":null}" + when: + def act = parser.parse(msg.bytes) + then: + act.type == ResponseWSParser.Type.RPC + act.id.asNumber() == 100L + act.error == null + new String(act.value) == "null" + } +} From b68da0f801be195e029c6477b9b68b924a48e1cb Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 20 Sep 2021 22:04:25 -0400 Subject: [PATCH 02/12] solution: WS-only Ethereum upstream --- .../dshackle/config/UpstreamsConfig.kt | 1 + .../dshackle/startup/ConfiguredUpstreams.kt | 36 ++++-- .../upstream/ethereum/EthereumRpcUpstream.kt | 15 +-- .../upstream/ethereum/EthereumUpstream.kt | 19 ++- .../upstream/ethereum/EthereumWsFactory.kt | 21 ++- .../upstream/ethereum/EthereumWsUpstream.kt | 120 ++++++++++++++++++ .../config/UpstreamsConfigReaderSpec.groovy | 26 ++++ src/test/resources/upstreams-ws-only.yaml | 13 ++ 8 files changed, 217 insertions(+), 34 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt create mode 100644 src/test/resources/upstreams-ws-only.yaml diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt index d0038558..4be0ba84 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt @@ -106,6 +106,7 @@ open class UpstreamsConfig { class EthereumConnection : RpcConnection() { var ws: WsEndpoint? = null + var preferHttp: Boolean = false } class BitcoinConnection : RpcConnection() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 488efe06..81e0d55f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -28,6 +28,7 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory +import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsUpstream import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcHttpClient import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -184,11 +185,6 @@ open class ConfiguredUpstreams( chain: Chain, options: UpstreamsConfig.Options) { val conn = config.connection!! - val directApi: Reader? = buildHttpClient(config) - if (directApi == null) { - log.warn("Upstream doesn't have API configuration") - return - } val urls = ArrayList() val methods = buildMethods(config, chain) @@ -209,13 +205,29 @@ open class ConfiguredUpstreams( } log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") - val ethereumUpstream = EthereumRpcUpstream( - config.id!!, - chain, directApi, wsFactoryApi, - options, config.role, - QuorumForLabels.QuorumItem(1, config.labels), - methods - ) + val ethereumUpstream = if (wsFactoryApi != null && !conn.preferHttp) { + EthereumWsUpstream( + config.id!!, + chain, wsFactoryApi, + options, config.role, + QuorumForLabels.QuorumItem(1, config.labels), + methods + ) + } else { + val directApi: Reader? = buildHttpClient(config) + if (directApi == null) { + log.warn("Upstream doesn't have API configuration") + return + } + EthereumRpcUpstream( + config.id!!, + chain, directApi, wsFactoryApi, + options, config.role, + QuorumForLabels.QuorumItem(1, config.labels), + methods + ) + } + ethereumUpstream.start() currentUpstreams.update(UpstreamChange(chain, ethereumUpstream, UpstreamChange.ChangeType.ADDED)) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index 1844fa49..2bae1815 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -38,11 +38,6 @@ open class EthereumRpcUpstream( private val head: Head = this.createHead() private var validatorSubscription: Disposable? = null - private val capabilities = if (options.providesBalance != false) { - setOf(Capability.RPC, Capability.BALANCE) - } else { - setOf(Capability.RPC) - } override fun setCaches(caches: Caches) { if (head is CachesEnabled) { @@ -79,7 +74,7 @@ open class EthereumRpcUpstream( open fun createHead(): Head { return if (ethereumWsFactory != null) { - val ws = ethereumWsFactory.create().apply { + val ws = ethereumWsFactory.create(null).apply { connect() } val wsHead = EthereumWsHead(ws).apply { @@ -108,14 +103,6 @@ open class EthereumRpcUpstream( return directReader } - override fun getLabels(): Collection { - return listOf(node.labels) - } - - override fun getCapabilities(): Set { - return capabilities - } - override fun isGrpc(): Boolean { return false } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index dbaaaf18..f629373e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -26,5 +26,20 @@ abstract class EthereumUpstream( options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?, - node: QuorumForLabels.QuorumItem? -) : DefaultUpstream(id, options, role, targets, node) \ No newline at end of file + private val node: QuorumForLabels.QuorumItem? +) : DefaultUpstream(id, options, role, targets, node) { + + private val capabilities = if (options.providesBalance != false) { + setOf(Capability.RPC, Capability.BALANCE) + } else { + setOf(Capability.RPC) + } + + override fun getCapabilities(): Set { + return capabilities + } + + override fun getLabels(): Collection { + return node?.let { listOf(it.labels) } ?: emptyList() + } +} \ No newline at end of file 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 37977e31..3ff2883e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser +import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics import io.emeraldpay.etherjar.rpc.json.BlockJson import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import io.netty.buffer.ByteBuf @@ -58,14 +59,15 @@ class EthereumWsFactory( var basicAuth: AuthConfig.ClientBasicAuth? = null - fun create(): EthereumWs { - return EthereumWs(uri, origin, basicAuth) + fun create(rpcMetrics: RpcMetrics?): EthereumWs { + return EthereumWs(uri, origin, basicAuth, rpcMetrics) } class EthereumWs( private val uri: URI, private val origin: URI, - private val basicAuth: AuthConfig.ClientBasicAuth? + private val basicAuth: AuthConfig.ClientBasicAuth?, + private val rpcMetrics: RpcMetrics? ) : AutoCloseable { companion object { @@ -255,12 +257,13 @@ class EthereumWsFactory( fun call(originalRequest: JsonRpcRequest): Mono { return Mono.fromCallable { + val startTime = System.nanoTime() // use an internal id sequence, to avoid id conflicts with user calls val internalId = sendIdSeq.getAndIncrement() val originalId = originalRequest.id - Tuples.of(originalRequest.copy(id = internalId), originalId) + Tuples.of(originalRequest.copy(id = internalId), originalId, startTime) }.flatMap { request -> - waitForResponse(request.t1, request.t2) + waitForResponse(request.t1, request.t2, request.t3) } } @@ -274,7 +277,7 @@ class EthereumWsFactory( } } - fun waitForResponse(request: JsonRpcRequest, originalId: Int): Mono { + fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono { val expectedId = request.id.toLong() return Mono.just(request) .flatMap { @@ -283,6 +286,12 @@ class EthereumWsFactory( .filter { resp -> resp.id.asNumber() == expectedId } .take(1) .singleOrEmpty() + .doOnNext { + rpcMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) + } + .doOnError { + rpcMetrics?.errors?.increment() + } .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt new file mode 100644 index 00000000..cd1163a8 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt @@ -0,0 +1,120 @@ +/** + * 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.upstream.ethereum + +import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.startup.QuorumForLabels +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.Upstream +import io.emeraldpay.dshackle.upstream.calls.CallMethods +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient +import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics +import io.emeraldpay.grpc.Chain +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.Metrics +import io.micrometer.core.instrument.Tag +import io.micrometer.core.instrument.Timer +import org.slf4j.LoggerFactory +import org.springframework.context.Lifecycle +import reactor.core.Disposable + +class EthereumWsUpstream( + id: String, + val chain: Chain, + ethereumWsFactory: EthereumWsFactory, + options: UpstreamsConfig.Options, + role: UpstreamsConfig.UpstreamRole, + node: QuorumForLabels.QuorumItem, + targets: CallMethods +) : EthereumUpstream(id, options, role, targets, node), Upstream, Lifecycle { + + companion object { + private val log = LoggerFactory.getLogger(EthereumWsUpstream::class.java) + } + + private val head: EthereumWsHead + private val connection: EthereumWsFactory.EthereumWs + private val api: JsonRpcWsClient + + private var validatorSubscription: Disposable? = null + + init { + val metricsTags = listOf( + Tag.of("upstream", id), + // UNSPECIFIED shouldn't happen too + Tag.of("chain", chain.chainCode) + ) + val metrics = RpcMetrics( + Timer.builder("upstream.ws.conn") + .description("Request time through a WebSocket JSON RPC connection") + .tags(metricsTags) + .publishPercentileHistogram() + .register(Metrics.globalRegistry), + Counter.builder("upstream.ws.err") + .description("Errors received on request through WebSocket JSON RPC connection") + .tags(metricsTags) + .register(Metrics.globalRegistry) + ) + + connection = ethereumWsFactory.create(metrics) + head = EthereumWsHead(connection) + api = JsonRpcWsClient(connection) + } + + override fun getHead(): Head { + return head + } + + override fun getApi(): Reader { + return api + } + + override fun isGrpc(): Boolean { + return false + } + + @Suppress("UNCHECKED_CAST") + override fun cast(selfType: Class): T { + if (!selfType.isAssignableFrom(this.javaClass)) { + throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") + } + return this as T + } + + override fun start() { + connection.connect() + head.start() + + log.debug("Start validation for upstream ${this.getId()}") + val validator = EthereumUpstreamValidator(this, getOptions()) + validatorSubscription = validator.start() + .subscribe(this::setStatus) + } + + override fun stop() { + validatorSubscription?.dispose() + validatorSubscription = null + head.stop() + connection.close() + } + + override fun isRunning(): Boolean { + return head.isRunning + } +} \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy index 5c1a553d..54608f81 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy @@ -74,6 +74,32 @@ class UpstreamsConfigReaderSpec extends Specification { } } + def "Parse websocket-only config"() { + setup: + def config = this.class.getClassLoader().getResourceAsStream("upstreams-ws-only.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" + } + } + } + } + + def "Parse bitcoin upstreams"() { setup: def config = this.class.getClassLoader().getResourceAsStream("upstreams-bitcoin.yaml") diff --git a/src/test/resources/upstreams-ws-only.yaml b/src/test/resources/upstreams-ws-only.yaml new file mode 100644 index 00000000..decb2ca2 --- /dev/null +++ b/src/test/resources/upstreams-ws-only.yaml @@ -0,0 +1,13 @@ +version: v1 + +upstreams: + - id: local + chain: ethereum + connection: + ethereum: + ws: + url: "ws://localhost:8546" + origin: "http://localhost" + basic-auth: + username: 9c199ad8f281f20154fc258fe41a6814 + password: 258fe4149c199ad8f2811a68f20154fc \ No newline at end of file From e40c62a728f28907ee1af06fbb5055ec50d822b2 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 20 Sep 2021 22:43:00 -0400 Subject: [PATCH 03/12] solution: check current height on connect --- .../upstream/ethereum/DefaultEthereumHead.kt | 39 +++++++++++++++++++ .../upstream/ethereum/EthereumRpcHead.kt | 25 +----------- .../upstream/ethereum/EthereumWsHead.kt | 9 ++++- 3 files changed, 48 insertions(+), 25 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt index ea912f82..3ccd4f8d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt @@ -15,9 +15,48 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.Defaults +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.upstream.AbstractHead import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.emeraldpay.etherjar.hex.HexQuantity +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono open class DefaultEthereumHead : Head, AbstractHead() { + companion object { + private val log = LoggerFactory.getLogger(DefaultEthereumHead::class.java) + } + + fun getLatestBlock(api: Reader): Mono { + return api.read(JsonRpcRequest("eth_blockNumber", emptyList())) + .subscribeOn(EthereumRpcHead.scheduler) + .timeout(Defaults.timeout, Mono.error(Exception("Block number not received"))) + .flatMap { + if (it.error != null) { + Mono.error(it.error.asException(null)) + } else { + val value = it.getResultAsProcessedString() + Mono.just(HexQuantity.from(value)) + } + } + .flatMap { + //fetching by Block Height here, critical to use the same upstream as in previous call, + //b/c different upstreams may have different blocks on the same height + api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false))) + .subscribeOn(EthereumRpcHead.scheduler) + .timeout(Defaults.timeout, Mono.error(Exception("Block data not received"))) + } + .map { + BlockContainer.fromEthereumJson(it.getResult()) + } + .onErrorResume { err -> + log.debug("Failed to fetch latest block: ${err.message}") + Mono.empty() + } + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt index d4b1971d..960e2266 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -49,30 +49,7 @@ class EthereumRpcHead( val base = Flux.interval(interval) .publishOn(scheduler) .flatMap { - api.read(JsonRpcRequest("eth_blockNumber", emptyList())) - .subscribeOn(scheduler) - .timeout(Defaults.timeout, Mono.error(Exception("Block number not received"))) - .flatMap { - if (it.error != null) { - Mono.error(it.error.asException(null)) - } else { - val value = it.getResultAsProcessedString() - Mono.just(HexQuantity.from(value)) - } - } - } - .flatMap { - //fetching by Block Height here, critical to use same upstream, - //different upstreams may have different blocks on the same height - api.read(JsonRpcRequest("eth_getBlockByNumber", listOf(it.toHex(), false))) - .subscribeOn(scheduler) - .timeout(Defaults.timeout, Mono.error(Exception("Block data not received"))) - } - .map { - BlockContainer.fromEthereumJson(it.getResult()) - } - .onErrorContinue { err, _ -> - log.debug("RPC error ${err.message}") + getLatestBlock(api) } refreshSubscription = super.follow(base) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index 3386611e..a24c0e0a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -16,9 +16,11 @@ */ package io.emeraldpay.dshackle.upstream.ethereum +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient import org.slf4j.LoggerFactory import org.springframework.context.Lifecycle import reactor.core.Disposable +import reactor.core.publisher.Flux class EthereumWsHead( private val ws: EthereumWsFactory.EthereumWs @@ -34,7 +36,12 @@ class EthereumWsHead( override fun start() { this.subscription?.dispose() - this.subscription = super.follow(ws.getBlocksFlux()) + val heads = Flux.merge( + // get the current block, not just wait for the next update + getLatestBlock(JsonRpcWsClient(ws)), + ws.getBlocksFlux() + ) + this.subscription = super.follow(heads) } override fun stop() { From 9a6b6128550536b22926fd33f3b1c8aefa908fe0 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 21 Sep 2021 20:00:16 -0400 Subject: [PATCH 04/12] solution: ensure it reconnects if WS connection closed or dropped --- build.gradle | 1 + .../upstream/ethereum/EthereumWsFactory.kt | 25 +++- .../dshackle/test/MockWSServer.groovy | 90 +++++++++++++++ .../ethereum/EthereumWsFactoryRealSpec.groovy | 108 ++++++++++++++++++ 4 files changed, 221 insertions(+), 3 deletions(-) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/test/MockWSServer.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy diff --git a/build.gradle b/build.gradle index 0e7f51fb..62a107db 100644 --- a/build.gradle +++ b/build.gradle @@ -132,6 +132,7 @@ dependencies { testImplementation "io.projectreactor:reactor-test:$reactorVersion" testImplementation 'org.objenesis:objenesis:3.1' testImplementation 'org.mock-server:mockserver-netty:5.11.2' + testImplementation "org.java-websocket:Java-WebSocket:1.5.1" testImplementation "nl.jqno.equalsverifier:equalsverifier:3.3" testImplementation "org.codehaus.groovy:groovy:${groovyVersion}" } 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 3ff2883e..6d24e20b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -77,6 +77,14 @@ class EthereumWsFactory( private const val START_REQUEST = "{\"jsonrpc\":\"2.0\", \"method\":\"eth_subscribe\", \"id\":\"blocks\", \"params\":[\"newHeads\"]}" } + var retryInterval = Defaults.retryConnection.seconds + set(value) { + if (retryInterval <= 0) { + throw IllegalArgumentException("Reconnect interval cannot be zero or less: $retryInterval") + } + field = value + } + private val parser = ResponseWSParser() private val blocks = Sinks @@ -103,15 +111,24 @@ class EthereumWsFactory( } private fun tryReconnectLater() { + if (!keepConnection) { + return + } + log.info("Reconnect to $uri in $retryInterval seconds...") Global.control.schedule( { connectInternal() }, - Defaults.retryConnection.seconds, TimeUnit.SECONDS) + retryInterval, TimeUnit.SECONDS) } private fun connectInternal() { log.info("Connecting to WebSocket: $uri") connection?.dispose() connection = HttpClient.create() + .doOnDisconnected { + if (keepConnection) { + tryReconnectLater() + } + } .doOnError( { _, t -> log.warn("Failed to connect to $uri. Error: ${t.message}") @@ -120,6 +137,7 @@ class EthereumWsFactory( }, { _, _ -> } ) + .headers { headers -> headers.add(HttpHeaderNames.ORIGIN, origin) basicAuth?.let { auth -> @@ -141,8 +159,9 @@ class EthereumWsFactory( .handle { inbound, outbound -> handle(inbound, outbound) } - .doOnError { - log.error("Failed to setup WS connection", it) + .onErrorResume { t -> + log.debug("Dropping WS connection to $uri. Error: ${t.message}") + Mono.empty() } .subscribe() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MockWSServer.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MockWSServer.groovy new file mode 100644 index 00000000..5a6ad76a --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MockWSServer.groovy @@ -0,0 +1,90 @@ +package io.emeraldpay.dshackle.test + +import com.fasterxml.jackson.databind.util.ByteBufferBackedInputStream +import org.java_websocket.WebSocket +import org.java_websocket.handshake.ClientHandshake +import org.java_websocket.server.WebSocketServer +import org.joda.time.format.DateTimeFormat + +import java.nio.ByteBuffer +import java.time.Instant +import java.time.LocalDate +import java.time.ZoneId +import java.time.format.DateTimeFormatter +import java.time.format.DateTimeFormatterBuilder + +class MockWSServer extends WebSocketServer { + + private def format = DateTimeFormatter.ofPattern("HH:mm:ss.SSS") + + List received = [] + private WebSocket conn + + private String next + + MockWSServer(int port) { + super(new InetSocketAddress("127.0.0.1", port)) + } + + void log(String msg) { + println(format.format(Instant.now().atZone(ZoneId.systemDefault())) + " MOCKWS: " + msg) + } + + void reply(String message) { + log(">> $message") + if (conn == null) { + println("MOCKWS: ERROR, no active connection") + } + conn.send(message) + } + + void onNextReply(String message) { + next = message + } + + @Override + void onOpen(WebSocket conn, ClientHandshake handshake) { + this.conn = conn + log("Opened connection from ${conn.remoteSocketAddress}") + } + + @Override + void onClose(WebSocket conn, int code, String reason, boolean remote) { + this.conn = null + log("Connection closed, code ${code} with msg '${reason}' ${remote ? 'by remote' : 'by server'}") + } + + @Override + void onMessage(WebSocket conn, String message) { + log("<< $message") + received.add(new ReceivedMessage(message)) + if (next != null) { + reply(next) + next = null + } + } + + @Override + void onMessage(WebSocket conn, ByteBuffer message) { + onMessage(conn, new ByteBufferBackedInputStream(message).text) + } + + @Override + void onError(WebSocket conn, Exception ex) { + log("ERROR, $ex.message") + received.add(new ReceivedMessage("Err: ${ex.message}")) + } + + @Override + void onStart() { + log("Server started") + } + + class ReceivedMessage { + final String value + + ReceivedMessage(String value) { + this.value = value + } + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy new file mode 100644 index 00000000..ca2a3b59 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy @@ -0,0 +1,108 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.test.MockWSServer +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest +import reactor.test.StepVerifier +import spock.lang.Shared +import spock.lang.Specification + +import java.time.Duration + +class EthereumWsFactoryRealSpec extends Specification { + + // needs large timeouts and sleep, especially on CI where it's much slower to run + static TIMEOUT = 15 + static SLEEP = 500 + + static int port = 19900 + new Random().nextInt(100) + @Shared + MockWSServer server + @Shared + EthereumWsFactory.EthereumWs conn + + def setup() { + port++ + server = new MockWSServer(port) + server.start() + Thread.sleep(SLEEP) + conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null) + } + + def cleanup() { + conn.close() + server.stop() + } + + def "Connects to server"() { + when: + conn.connect() + Thread.sleep(SLEEP) + println("verify....") + def act = server.received + then: + act.size() > 0 + act[0].value.contains("\"method\":\"eth_subscribe\"") + act[0].value.contains("\"params\":[\"newHeads\"]") + } + + def "Makes RPC request"() { + when: + conn.connect() + def resp = conn.call(new JsonRpcRequest("foo_bar", [])) + then: + StepVerifier.create(resp) + .then { + server.reply('{"jsonrpc":"2.0", "id":100, "result": "baz"}') + } + .expectNextMatches { + it.hasResult() && it.resultAsProcessedString == "baz" + } + .expectComplete() + .verify(Duration.ofSeconds(3)) + + when: + Thread.sleep(SLEEP) + def act = server.received + then: + act.size() == 2 + act[1].value.contains("\"method\":\"foo_bar\"") + } + + def "Reconnects after server disconnect"() { + when: + conn.connect() + conn.retryInterval = 2 + Thread.sleep(SLEEP) + server.stop() + Thread.sleep(SLEEP) + server = new MockWSServer(port) + server.start() + def resp = conn.call(new JsonRpcRequest("foo_bar", [])) + // reconnects in 2 seconds, give 1 extra + Thread.sleep(3_000) + def act = server.received + + then: + act.size() > 0 + act[0].value.contains("\"method\":\"eth_subscribe\"") + act[0].value.contains("\"params\":[\"newHeads\"]") + } + + def "Try to connects to server until it's available"() { + when: + server.stop() + Thread.sleep(SLEEP) + conn.retryInterval = 1 + conn.connect() + Thread.sleep(3_000) + server = new MockWSServer(port) + server.start() + Thread.sleep(2_000) + def act = server.received + then: + act.size() > 0 + act[0].value.contains("\"method\":\"eth_subscribe\"") + act[0].value.contains("\"params\":[\"newHeads\"]") + } + +} From 613c895d75fd2b959df7169d93f828da1c698d41 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 21 Sep 2021 20:28:19 -0400 Subject: [PATCH 05/12] problem: may stick with reconnect if multiple errors happened at the same time --- .../dshackle/upstream/ethereum/EthereumWsFactory.kt | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) 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 6d24e20b..0321155f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -49,6 +49,7 @@ import java.time.Duration import java.util.* import java.util.concurrent.Executors import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger @@ -103,6 +104,7 @@ class EthereumWsFactory( private val sendExecutor = Executors.newSingleThreadExecutor() private var keepConnection = true private var connection: Disposable? = null + private val reconnecting = AtomicBoolean(false) fun connect() { if (keepConnection) { @@ -114,9 +116,16 @@ class EthereumWsFactory( if (!keepConnection) { return } + val alreadyReconnecting = reconnecting.getAndSet(true) + if (alreadyReconnecting) { + return + } log.info("Reconnect to $uri in $retryInterval seconds...") Global.control.schedule( - { connectInternal() }, + { + reconnecting.set(false) + connectInternal() + }, retryInterval, TimeUnit.SECONDS) } @@ -125,6 +134,7 @@ class EthereumWsFactory( connection?.dispose() connection = HttpClient.create() .doOnDisconnected { + log.info("Disconnected from $uri") if (keepConnection) { tryReconnectLater() } From e44b2c129da1374029b2fefc954e4ba243662f19 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 21 Sep 2021 20:33:53 -0400 Subject: [PATCH 06/12] problem: doesn't execute calls after reconnect --- .../upstream/ethereum/EthereumWsFactory.kt | 9 ++++++- .../ethereum/EthereumWsFactoryRealSpec.groovy | 24 +++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) 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 0321155f..28785e8a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -92,7 +92,7 @@ class EthereumWsFactory( .many() .multicast() .directBestEffort() - private val rpcSend = Sinks + private var rpcSend = Sinks .many() .unicast() .onBackpressureBuffer() @@ -120,6 +120,13 @@ class EthereumWsFactory( if (alreadyReconnecting) { return } + // rpcSend is already CANCELLED, since the subscription owned by the previous connection is gone + // so we need to create a new Sink. Emit Complete is probably useless, and just in case + rpcSend.tryEmitComplete() + rpcSend = Sinks + .many() + .unicast() + .onBackpressureBuffer() log.info("Reconnect to $uri in $retryInterval seconds...") Global.control.schedule( { diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy index ca2a3b59..bf66d4e9 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy @@ -105,4 +105,28 @@ class EthereumWsFactoryRealSpec extends Specification { act[0].value.contains("\"params\":[\"newHeads\"]") } + def "Call after reconnect"() { + when: + conn.connect() + conn.retryInterval = 2 + Thread.sleep(SLEEP) + server.stop() + Thread.sleep(SLEEP) + server = new MockWSServer(port) + server.start() + // reconnects in 2 seconds, give 1 extra + Thread.sleep(3_000) + + def resp = conn.call(new JsonRpcRequest("foo_bar", [])) + then: + StepVerifier.create(resp) + .then { + server.reply('{"jsonrpc":"2.0", "id":100, "result": "baz"}') + } + .expectNextMatches { + it.hasResult() && it.resultAsProcessedString == "baz" + } + .expectComplete() + .verify(Duration.ofSeconds(3)) + } } From 16e589ab69588a768baa65df357cb9b8e002f4d0 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 21 Sep 2021 21:07:28 -0400 Subject: [PATCH 07/12] problem: drops connection because of large frames --- .../dshackle/upstream/ethereum/EthereumWsFactory.kt | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) 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 28785e8a..6e70e63a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -170,6 +170,11 @@ 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) .build() ) .uri(uri) @@ -184,9 +189,9 @@ class EthereumWsFactory( } fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher { - val consumer = inbound.aggregateFrames() - // accept up to 1Mb messages - .aggregateFrames(16 * 65_536) + val consumer = inbound + // Accept up to 15Mb messages, same config is used by Geth + .aggregateFrames(15 * 1024 * 1024) .receiveFrames() .map { ByteBufInputStream(it.content()).readAllBytes() } .flatMap { @@ -203,7 +208,7 @@ class EthereumWsFactory( } } .onErrorResume { t -> - log.warn("Connection dropped to $uri. Error: ${t.message}") + log.warn("Connection dropped to $uri. Error: ${t.message}", t) // going to try to reconnect later tryReconnectLater() // completes current outbound flow From 37e3d4e25246a1ed97d30900576cedcd21cb3829 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 21 Sep 2021 21:41:44 -0400 Subject: [PATCH 08/12] problem: waits to long to reconnect (10 seconds) solution: use exp backoff, from 100ms to 1 minute --- .../upstream/ethereum/EthereumWsFactory.kt | 34 +++++++++++++------ 1 file changed, 24 insertions(+), 10 deletions(-) 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 6e70e63a..19b06dae 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -33,6 +33,10 @@ import io.netty.buffer.Unpooled import io.netty.handler.codec.http.HttpHeaderNames import org.reactivestreams.Publisher import org.slf4j.LoggerFactory +import org.springframework.util.backoff.BackOff +import org.springframework.util.backoff.BackOffExecution +import org.springframework.util.backoff.ExponentialBackOff +import org.springframework.util.backoff.FixedBackOff import reactor.core.Disposable import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -78,13 +82,11 @@ class EthereumWsFactory( private const val START_REQUEST = "{\"jsonrpc\":\"2.0\", \"method\":\"eth_subscribe\", \"id\":\"blocks\", \"params\":[\"newHeads\"]}" } - var retryInterval = Defaults.retryConnection.seconds - set(value) { - if (retryInterval <= 0) { - throw IllegalArgumentException("Reconnect interval cannot be zero or less: $retryInterval") - } - field = value - } + private var reconnectBackoff: BackOff = ExponentialBackOff().also { + it.initialInterval = Duration.ofMillis(100).toMillis() + it.maxInterval = Duration.ofMinutes(1).toMillis() + } + private var currentBackOff = reconnectBackoff.start() private val parser = ResponseWSParser() @@ -106,6 +108,11 @@ class EthereumWsFactory( private var connection: Disposable? = null private val reconnecting = AtomicBoolean(false) + fun setReconnectIntervalSeconds(value: Long) { + reconnectBackoff = FixedBackOff(value * 1000, FixedBackOff.UNLIMITED_ATTEMPTS) + currentBackOff = reconnectBackoff.start() + } + fun connect() { if (keepConnection) { connectInternal() @@ -127,13 +134,18 @@ class EthereumWsFactory( .many() .unicast() .onBackpressureBuffer() - log.info("Reconnect to $uri in $retryInterval seconds...") + val retryInterval = currentBackOff.nextBackOff() + if (retryInterval == BackOffExecution.STOP) { + log.warn("Reconnect backoff exhausted. Permanently closing the connection") + return + } + log.info("Reconnect to $uri in ${retryInterval}ms...") Global.control.schedule( { reconnecting.set(false) connectInternal() }, - retryInterval, TimeUnit.SECONDS) + retryInterval, TimeUnit.MILLISECONDS) } private fun connectInternal() { @@ -154,7 +166,6 @@ class EthereumWsFactory( }, { _, _ -> } ) - .headers { headers -> headers.add(HttpHeaderNames.ORIGIN, origin) basicAuth?.let { auth -> @@ -189,6 +200,9 @@ class EthereumWsFactory( } fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher { + //restart backoff after connection + currentBackOff = reconnectBackoff.start() + val consumer = inbound // Accept up to 15Mb messages, same config is used by Geth .aggregateFrames(15 * 1024 * 1024) From 5586a95063c597faa670ad3d81a8a94807123933 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 21 Sep 2021 21:54:57 -0400 Subject: [PATCH 09/12] problem: unit test rel: [37e3d4e] --- .../upstream/ethereum/EthereumWsFactoryRealSpec.groovy | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy index bf66d4e9..a884b1de 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy @@ -71,7 +71,7 @@ class EthereumWsFactoryRealSpec extends Specification { def "Reconnects after server disconnect"() { when: conn.connect() - conn.retryInterval = 2 + conn.reconnectIntervalSeconds = 2 Thread.sleep(SLEEP) server.stop() Thread.sleep(SLEEP) @@ -92,7 +92,7 @@ class EthereumWsFactoryRealSpec extends Specification { when: server.stop() Thread.sleep(SLEEP) - conn.retryInterval = 1 + conn.reconnectIntervalSeconds = 1 conn.connect() Thread.sleep(3_000) server = new MockWSServer(port) @@ -108,7 +108,7 @@ class EthereumWsFactoryRealSpec extends Specification { def "Call after reconnect"() { when: conn.connect() - conn.retryInterval = 2 + conn.reconnectIntervalSeconds = 2 Thread.sleep(SLEEP) server.stop() Thread.sleep(SLEEP) From 1f864bdbe31e297a7e41d3dc7c11cd79e2e6ee34 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Tue, 21 Sep 2021 22:06:44 -0400 Subject: [PATCH 10/12] problem: too slow to validate WS on CI --- .github/workflows/test.yaml | 3 +++ .../io/emeraldpay/dshackle/test/MockWSServer.groovy | 2 +- .../ethereum/EthereumWsFactoryRealSpec.groovy | 11 +++++++---- 3 files changed, 11 insertions(+), 5 deletions(-) diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 8ba13463..f19f1627 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -32,6 +32,8 @@ jobs: uses: eskatos/gradle-command-action@v1 with: arguments: check + env: + CI: true - name: Upload Coverage Report uses: codecov/codecov-action@v1 @@ -64,6 +66,7 @@ jobs: with: arguments: check env: + CI: true DSHACKLE_TEST_ENABLED: redis REDIS_HOST: redis REDIS_PORT: 6379 \ No newline at end of file diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MockWSServer.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MockWSServer.groovy index 5a6ad76a..65f6ba4d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MockWSServer.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MockWSServer.groovy @@ -33,7 +33,7 @@ class MockWSServer extends WebSocketServer { void reply(String message) { log(">> $message") if (conn == null) { - println("MOCKWS: ERROR, no active connection") + log("MOCKWS: ERROR, no active connection") } conn.send(message) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy index a884b1de..8edfc3b2 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy @@ -10,8 +10,6 @@ import java.time.Duration class EthereumWsFactoryRealSpec extends Specification { - // needs large timeouts and sleep, especially on CI where it's much slower to run - static TIMEOUT = 15 static SLEEP = 500 static int port = 19900 + new Random().nextInt(100) @@ -21,6 +19,11 @@ class EthereumWsFactoryRealSpec extends Specification { EthereumWsFactory.EthereumWs conn def setup() { + if (System.getenv("CI") == "true") { + println("RUN IN CI ENVIRONMENT") + // needs large timeouts on CI where it's much slower to run + SLEEP = 1500 + } port++ server = new MockWSServer(port) server.start() @@ -52,7 +55,7 @@ class EthereumWsFactoryRealSpec extends Specification { then: StepVerifier.create(resp) .then { - server.reply('{"jsonrpc":"2.0", "id":100, "result": "baz"}') + server.onNextReply('{"jsonrpc":"2.0", "id":100, "result": "baz"}') } .expectNextMatches { it.hasResult() && it.resultAsProcessedString == "baz" @@ -121,7 +124,7 @@ class EthereumWsFactoryRealSpec extends Specification { then: StepVerifier.create(resp) .then { - server.reply('{"jsonrpc":"2.0", "id":100, "result": "baz"}') + server.onNextReply('{"jsonrpc":"2.0", "id":100, "result": "baz"}') } .expectNextMatches { it.hasResult() && it.resultAsProcessedString == "baz" From bd8c0f7c19df3580eb25e587c12568e1148e12fa Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 22 Sep 2021 17:08:43 -0400 Subject: [PATCH 11/12] solution: mark Upstream status right after WS connect/disconnect --- .../dshackle/upstream/DefaultUpstream.kt | 2 +- .../upstream/ethereum/EthereumRpcUpstream.kt | 3 +- .../ethereum/EthereumUpstreamValidator.kt | 4 +-- .../upstream/ethereum/EthereumWsFactory.kt | 15 +++++++-- .../upstream/ethereum/EthereumWsUpstream.kt | 6 ++-- .../ethereum/EthereumWsFactoryRealSpec.groovy | 31 ++++++++++++++++++- .../ethereum/EthereumWsFactorySpec.groovy | 8 ++--- 7 files changed, 55 insertions(+), 14 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 86d54225..6646d85b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -61,7 +61,7 @@ abstract class DefaultUpstream( return status.get().status } - fun setStatus(avail: UpstreamAvailability) { + open fun setStatus(avail: UpstreamAvailability) { status.updateAndGet { curr -> Status(curr.lag, avail, statusByLag(curr.lag, avail)) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index 2bae1815..b354abc9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -74,7 +74,8 @@ open class EthereumRpcUpstream( open fun createHead(): Head { return if (ethereumWsFactory != null) { - val ws = ethereumWsFactory.create(null).apply { + // do not set upstream to the WS, since it doesn't control the RPC upstream + val ws = ethereumWsFactory.create(null, null, null).apply { connect() } val wsHead = EthereumWsHead(ws).apply { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt index 99d326b1..a968753a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt @@ -32,7 +32,7 @@ import reactor.core.scheduler.Schedulers import java.time.Duration import java.util.concurrent.Executors -class EthereumUpstreamValidator( +open class EthereumUpstreamValidator( private val upstream: EthereumUpstream, private val options: UpstreamsConfig.Options ) { @@ -43,7 +43,7 @@ class EthereumUpstreamValidator( private val objectMapper: ObjectMapper = Global.objectMapper - fun validate(): Mono { + open fun validate(): Mono { return upstream .getApi() .read(JsonRpcRequest("eth_syncing", listOf())) 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 19b06dae..60b98efc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -21,6 +21,8 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser @@ -64,15 +66,17 @@ class EthereumWsFactory( var basicAuth: AuthConfig.ClientBasicAuth? = null - fun create(rpcMetrics: RpcMetrics?): EthereumWs { - return EthereumWs(uri, origin, basicAuth, rpcMetrics) + fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?, rpcMetrics: RpcMetrics?): EthereumWs { + return EthereumWs(uri, origin, basicAuth, rpcMetrics, upstream, validator) } class EthereumWs( private val uri: URI, private val origin: URI, private val basicAuth: AuthConfig.ClientBasicAuth?, - private val rpcMetrics: RpcMetrics? + private val rpcMetrics: RpcMetrics?, + private val upstream: DefaultUpstream?, + private val validator: EthereumUpstreamValidator? ) : AutoCloseable { companion object { @@ -154,6 +158,8 @@ class EthereumWsFactory( connection = HttpClient.create() .doOnDisconnected { log.info("Disconnected from $uri") + // mark upstream as UNAVAIL + upstream?.setStatus(UpstreamAvailability.UNAVAILABLE) if (keepConnection) { tryReconnectLater() } @@ -203,6 +209,9 @@ class EthereumWsFactory( //restart backoff after connection currentBackOff = reconnectBackoff.start() + //validate the connection, it can also be UNAVAIL if market as such after disconnect + validator?.validate() + val consumer = inbound // Accept up to 15Mb messages, same config is used by Geth .aggregateFrames(15 * 1024 * 1024) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt index cd1163a8..7046be7b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt @@ -53,6 +53,7 @@ class EthereumWsUpstream( private val api: JsonRpcWsClient private var validatorSubscription: Disposable? = null + private val validator: EthereumUpstreamValidator init { val metricsTags = listOf( @@ -72,7 +73,9 @@ class EthereumWsUpstream( .register(Metrics.globalRegistry) ) - connection = ethereumWsFactory.create(metrics) + validator = EthereumUpstreamValidator(this, getOptions()) + + connection = ethereumWsFactory.create(this, validator, metrics) head = EthereumWsHead(connection) api = JsonRpcWsClient(connection) } @@ -102,7 +105,6 @@ class EthereumWsUpstream( head.start() log.debug("Start validation for upstream ${this.getId()}") - val validator = EthereumUpstreamValidator(this, getOptions()) validatorSubscription = validator.start() .subscribe(this::setStatus) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy index 8edfc3b2..069c7a05 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy @@ -1,6 +1,8 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.test.MockWSServer +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import reactor.test.StepVerifier import spock.lang.Shared @@ -28,7 +30,7 @@ class EthereumWsFactoryRealSpec extends Specification { server = new MockWSServer(port) server.start() Thread.sleep(SLEEP) - conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null) + conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null, null) } def cleanup() { @@ -91,6 +93,33 @@ class EthereumWsFactoryRealSpec extends Specification { act[0].value.contains("\"params\":[\"newHeads\"]") } + def "Gets UNAVAIL status right after disconnect"() { + setup: + def up = Mock(DefaultUpstream) + conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null, null) + when: + conn.connect() + conn.reconnectIntervalSeconds = 10 + Thread.sleep(SLEEP) + server.stop() + Thread.sleep(100) + + then: + 1 * up.setStatus(UpstreamAvailability.UNAVAILABLE) + } + + def "Validates after connect"() { + setup: + def validator = Mock(EthereumUpstreamValidator) + conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator, null) + when: + conn.connect() + Thread.sleep(100) + + then: + 1 * validator.validate() + } + def "Try to connects to server until it's available"() { when: server.stop() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy index f5415999..6da34f11 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy @@ -53,7 +53,7 @@ class EthereumWsFactorySpec extends Specification { def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create() + def ws = wsf.create(null, null, null) apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block) @@ -74,7 +74,7 @@ class EthereumWsFactorySpec extends Specification { def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create() + def ws = wsf.create(null, null, null) def tx = new TransactionJson().tap { hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") @@ -99,7 +99,7 @@ class EthereumWsFactorySpec extends Specification { def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create() + def ws = wsf.create(null, null, null) apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null) @@ -122,7 +122,7 @@ class EthereumWsFactorySpec extends Specification { def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create() + def ws = wsf.create(null, null, null) apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test")) From ebef8cf27ac63c9743c550a464f3719b2ef30fd8 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 22 Sep 2021 19:35:14 -0400 Subject: [PATCH 12/12] 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