StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

more reliabale ws connection backoff

Browse Source
This commit is contained in:
terminal
2022-09-06 13:41:30 +03:00
parent 2412132b08
commit b11bb5e5a3
1 changed files with 10 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt
View File

@@ -58,6 +58,7 @@ import java.time.Duration
import java.time.Instant import java.time.Instant
import java.util.Base64 import java.util.Base64
import java.util.concurrent.Executors import java.util.concurrent.Executors
import java.util.concurrent.ScheduledFuture
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicBoolean 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 // > 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) // 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 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 // 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 // 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 parser = ResponseWSParser()
private val resetBackoffExecutor = Executors.newScheduledThreadPool(2)
private var resetBackoffTask: ScheduledFuture<Unit>? = null
private val blocks = Sinks private val blocks = Sinks
.many() .many()
.multicast() .multicast()
@@ -147,6 +152,7 @@ open class WsConnection(
if (alreadyReconnecting) { if (alreadyReconnecting) {
return return
} }
// rpcSend is already CANCELLED, since the subscription owned by the previous connection is gone // 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 // so we need to create a new Sink. Emit Complete is probably useless, and just in case
rpcSend.tryEmitComplete() rpcSend.tryEmitComplete()
@@ -154,7 +160,11 @@ open class WsConnection(
.many() .many()
.unicast() .unicast()
.onBackpressureBuffer<JsonRpcRequest>() .onBackpressureBuffer<JsonRpcRequest>()
resetBackoffTask?.cancel(false)
val retryInterval = currentBackOff.nextBackOff() val retryInterval = currentBackOff.nextBackOff()
resetBackoffTask = resetBackoffExecutor.schedule<Unit>({
currentBackOff = reconnectBackoff.start()
}, RESET_BACKOFF_TIMEOUT + retryInterval, TimeUnit.MILLISECONDS)
if (retryInterval == BackOffExecution.STOP) { if (retryInterval == BackOffExecution.STOP) {
log.warn("Reconnect backoff exhausted. Permanently closing the connection") log.warn("Reconnect backoff exhausted. Permanently closing the connection")
return return
@@ -225,7 +235,6 @@ open class WsConnection(
} }
fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher<Void> { fun handle(inbound: WebsocketInbound, outbound: WebsocketOutbound): Publisher<Void> {
var read = false
val consumer = inbound val consumer = inbound
.aggregateFrames(msgSizeLimit) .aggregateFrames(msgSizeLimit)
.receiveFrames() .receiveFrames()
@@ -233,16 +242,6 @@ open class WsConnection(
.flatMap { .flatMap {
try { try {
val msg = parser.parse(it) 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) { if (msg.type == ResponseWSParser.Type.SUBSCRIPTION) {
onSubscription(msg) onSubscription(msg)
} else { } else {