From b11bb5e5a3eb2e274aacd4fa133c8cd48271938c Mon Sep 17 00:00:00 2001
From: terminal <terminal2010@gmail.com>
Date: Tue, 6 Sep 2022 13:41:30 +0300
Subject: [PATCH] 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<Unit>? = 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<JsonRpcRequest>()
+        resetBackoffTask?.cancel(false)
         val retryInterval = currentBackOff.nextBackOff()
+        resetBackoffTask = resetBackoffExecutor.schedule<Unit>({
+            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<Void> {
-        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 {