From a92b156e8fb463f2384020d0b994a6673d05a54c Mon Sep 17 00:00:00 2001 From: terminal Date: Tue, 6 Sep 2022 12:19:06 +0300 Subject: [PATCH 1/4] fix formatting --- .../emeraldpay/dshackle/upstream/ethereum/WsConnection.kt | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index 197fd056..a6befbc0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt @@ -165,7 +165,8 @@ open class WsConnection( reconnecting.set(false) connectInternal() }, - retryInterval, TimeUnit.MILLISECONDS + retryInterval, + TimeUnit.MILLISECONDS ) } @@ -298,7 +299,10 @@ open class WsConnection( fun onRpc(msg: ResponseWSParser.WsResponse): Mono { return if (msg.id.isNumber()) { val resp = JsonRpcResponse( - msg.value, msg.error, msg.id, null + msg.value, + msg.error, + msg.id, + null ) Mono.fromCallable { val status = rpcReceive.tryEmitNext(resp) From f791aa98ff909cd32ca7b69529ab71985057f1ed Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 25 Aug 2022 17:08:53 -0400 Subject: [PATCH 2/4] problem: reconnects too fast to a faulty WS connection --- .../dshackle/upstream/ethereum/WsConnection.kt | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index a6befbc0..54ac488f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt @@ -225,12 +225,18 @@ open class WsConnection( } fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher { - // restart backoff after connection - currentBackOff = reconnectBackoff.start() - + var read = false val consumer = inbound .aggregateFrames(msgSizeLimit) .receiveFrames() + .doOnNext { + if (!read) { + // restart backoff only after a successful read from the connection, + // otherwise it may restart it even if the connection is faulty + currentBackOff = reconnectBackoff.start() + read = true + } + } .map { ByteBufInputStream(it.content()).readAllBytes() } .flatMap { try { From 2412132b08c6c7df115d38901d65ec36abe7fbef Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 25 Aug 2022 19:18:19 -0400 Subject: [PATCH 3/4] problem: reconnects WS too often when the only response is an error --- .../upstream/ethereum/WsConnection.kt | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index 54ac488f..eb67320d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt @@ -96,7 +96,7 @@ open class WsConnection( private var reconnectBackoff: BackOff = ExponentialBackOff().also { it.initialInterval = Duration.ofMillis(100).toMillis() - it.maxInterval = Duration.ofMinutes(1).toMillis() + it.maxInterval = Duration.ofMinutes(5).toMillis() } private var currentBackOff = reconnectBackoff.start() @@ -229,18 +229,20 @@ open class WsConnection( val consumer = inbound .aggregateFrames(msgSizeLimit) .receiveFrames() - .doOnNext { - if (!read) { - // restart backoff only after a successful read from the connection, - // otherwise it may restart it even if the connection is faulty - currentBackOff = reconnectBackoff.start() - read = true - } - } .map { ByteBufInputStream(it.content()).readAllBytes() } .flatMap { try { val msg = parser.parse(it) + if (!read) { + if (msg.error != null) { + log.warn("Received error ${msg.error.code} from $uri: ${msg.error.message}") + } else { + // restart backoff only after a successful read from the connection, + // otherwise it may restart it even if the connection is faulty or responds only with error + currentBackOff = reconnectBackoff.start() + read = true + } + } if (msg.type == ResponseWSParser.Type.SUBSCRIPTION) { onSubscription(msg) } else { From b11bb5e5a3eb2e274aacd4fa133c8cd48271938c Mon Sep 17 00:00:00 2001 From: terminal Date: Tue, 6 Sep 2022 13:41:30 +0300 Subject: [PATCH 4/4] more reliabale ws connection backoff --- .../upstream/ethereum/WsConnection.kt | 21 +++++++++---------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index eb67320d..cbf8c6cf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt @@ -58,6 +58,7 @@ import java.time.Duration import java.time.Instant import java.util.Base64 import java.util.concurrent.Executors +import java.util.concurrent.ScheduledFuture import java.util.concurrent.TimeUnit import java.util.concurrent.TimeoutException import java.util.concurrent.atomic.AtomicBoolean @@ -85,6 +86,7 @@ open class WsConnection( // > io.netty.handler.codec.http.websocketx.CorruptedWebSocketFrameException: Max frame length of 65536 has been exceeded // It's unclear what is the right limit here, but 5mb seems to be working (1mb isn't always working) private const val DEFAULT_FRAME_SIZE = 5 * 1024 * 1024 + private const val RESET_BACKOFF_TIMEOUT = 10 * 1000 // 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 should be a right value @@ -102,6 +104,9 @@ open class WsConnection( private val parser = ResponseWSParser() + private val resetBackoffExecutor = Executors.newScheduledThreadPool(2) + private var resetBackoffTask: ScheduledFuture? = null + private val blocks = Sinks .many() .multicast() @@ -147,6 +152,7 @@ open class WsConnection( 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() @@ -154,7 +160,11 @@ open class WsConnection( .many() .unicast() .onBackpressureBuffer() + resetBackoffTask?.cancel(false) val retryInterval = currentBackOff.nextBackOff() + resetBackoffTask = resetBackoffExecutor.schedule({ + currentBackOff = reconnectBackoff.start() + }, RESET_BACKOFF_TIMEOUT + retryInterval, TimeUnit.MILLISECONDS) if (retryInterval == BackOffExecution.STOP) { log.warn("Reconnect backoff exhausted. Permanently closing the connection") return @@ -225,7 +235,6 @@ open class WsConnection( } fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher { - var read = false val consumer = inbound .aggregateFrames(msgSizeLimit) .receiveFrames() @@ -233,16 +242,6 @@ open class WsConnection( .flatMap { try { val msg = parser.parse(it) - if (!read) { - if (msg.error != null) { - log.warn("Received error ${msg.error.code} from $uri: ${msg.error.message}") - } else { - // restart backoff only after a successful read from the connection, - // otherwise it may restart it even if the connection is faulty or responds only with error - currentBackOff = reconnectBackoff.start() - read = true - } - } if (msg.type == ResponseWSParser.Type.SUBSCRIPTION) { onSubscription(msg) } else {