mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-09 16:14:40 +00:00
fix: don't reset relay reconnect backoff on momentary connections
A relay that accepts the WebSocket handshake and then immediately resets the connection (e.g. essayist.decentnewsroom.com) defeated the exponential reconnect backoff.
This commit is contained in:
davotoula
+28
-5
@@ -44,8 +44,11 @@ import kotlin.coroutines.cancellation.CancellationException
|
||||
* @property listener Interface to notify the application of relay events and errors.
|
||||
*
|
||||
* Reconnection Strategy:
|
||||
* - Uses exponential backoff to retry connections, starting with [DELAY_TO_RECONNECT_IN_SECS] (500ms).
|
||||
* - Uses exponential backoff to retry connections, starting with [DELAY_TO_RECONNECT_IN_SECS].
|
||||
* - Doubles the delay between reconnection attempts in case of failure.
|
||||
* - The backoff only resets after a connection stays open for at least
|
||||
* [STABLE_CONNECTION_IN_SECS], so relays that accept the handshake but
|
||||
* immediately drop the socket keep backing off instead of reconnecting in a loop.
|
||||
*
|
||||
* Message Handling:
|
||||
* - Processes relay messages (e.g., `EVENT`, `EOSE`, `OK`, `AUTH`) and delegates to appropriate callbacks.
|
||||
@@ -56,10 +59,17 @@ open class BasicRelayClient(
|
||||
override val url: NormalizedRelayUrl,
|
||||
val socketBuilder: WebsocketBuilder,
|
||||
val listener: RelayConnectionListener,
|
||||
val nowInSeconds: () -> Long = TimeUtils::now,
|
||||
) : IRelayClient {
|
||||
companion object {
|
||||
// minimum wait time to reconnect: 1 second
|
||||
const val DELAY_TO_RECONNECT_IN_SECS = 1
|
||||
|
||||
// a connection must survive this long before a disconnect resets the
|
||||
// reconnect backoff. Relays that accept the handshake and then
|
||||
// immediately drop the socket would otherwise reset the backoff on
|
||||
// every onOpen and reconnect in a tight ~3s loop forever.
|
||||
const val STABLE_CONNECTION_IN_SECS = TimeUtils.ONE_MINUTE
|
||||
}
|
||||
|
||||
private var socket: WebSocket? = null
|
||||
@@ -74,6 +84,10 @@ open class BasicRelayClient(
|
||||
private var lastConnectTentativeInSeconds: Long = 0L // the beginning of time.
|
||||
private var delayToConnectInSeconds = DELAY_TO_RECONNECT_IN_SECS
|
||||
|
||||
// when the current connection became ready; used to decide if the
|
||||
// connection was stable enough to reset the backoff on disconnect.
|
||||
private var connectedAtInSeconds: Long = 0L
|
||||
|
||||
// Makes sure only one socket is open for each url
|
||||
private var connectingMutex = AtomicBoolean(false)
|
||||
|
||||
@@ -97,7 +111,7 @@ open class BasicRelayClient(
|
||||
|
||||
listener.onConnecting(this)
|
||||
|
||||
lastConnectTentativeInSeconds = TimeUtils.now()
|
||||
lastConnectTentativeInSeconds = nowInSeconds()
|
||||
|
||||
socket = socketBuilder.build(url, MyWebsocketListener())
|
||||
socket?.connect()
|
||||
@@ -193,12 +207,21 @@ open class BasicRelayClient(
|
||||
fun markConnectionAsReady(usingCompression: Boolean) {
|
||||
this.isReady = true
|
||||
this.usingCompression = usingCompression
|
||||
this.connectedAtInSeconds = nowInSeconds()
|
||||
|
||||
// resets any extra delays added during on offline state
|
||||
this.delayToConnectInSeconds = DELAY_TO_RECONNECT_IN_SECS
|
||||
// The backoff delay is NOT reset here: a relay that accepts the
|
||||
// handshake and then immediately fails would defeat the exponential
|
||||
// backoff on every cycle. It resets in markConnectionAsClosed once
|
||||
// the session proves stable (see STABLE_CONNECTION_IN_SECS).
|
||||
}
|
||||
|
||||
fun markConnectionAsClosed() {
|
||||
// resets any extra delays added while offline, but only if the
|
||||
// session was stable; flapping relays keep their growing backoff.
|
||||
if (isReady && nowInSeconds() - connectedAtInSeconds >= STABLE_CONNECTION_IN_SECS) {
|
||||
this.delayToConnectInSeconds = DELAY_TO_RECONNECT_IN_SECS
|
||||
}
|
||||
|
||||
this.socket = null
|
||||
this.isReady = false
|
||||
this.usingCompression = false
|
||||
@@ -232,7 +255,7 @@ open class BasicRelayClient(
|
||||
}
|
||||
|
||||
// waits 60 seconds to reconnect after disconnected.
|
||||
if (ignoreRetryDelays || TimeUtils.now() > lastConnectTentativeInSeconds + delayToConnectInSeconds) {
|
||||
if (ignoreRetryDelays || nowInSeconds() > lastConnectTentativeInSeconds + delayToConnectInSeconds) {
|
||||
upRelayDelayToConnect()
|
||||
connect()
|
||||
}
|
||||
|
||||
+185
@@ -0,0 +1,185 @@
|
||||
/*
|
||||
* Copyright (c) 2025 Vitor Pamplona
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person obtaining a copy of
|
||||
* this software and associated documentation files (the "Software"), to deal in
|
||||
* the Software without restriction, including without limitation the rights to use,
|
||||
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
|
||||
* Software, and to permit persons to whom the Software is furnished to do so,
|
||||
* subject to the following conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be included in all
|
||||
* copies or substantial portions of the Software.
|
||||
*
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
|
||||
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
|
||||
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
|
||||
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
||||
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
*/
|
||||
package com.vitorpamplona.quartz.nip01Core.relay.client.single.basic
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebsocketBuilder
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* Reproduces the reconnect storm from the 2026-06-12 benchmark sweep: a relay
|
||||
* that completes the WebSocket handshake and then immediately resets the
|
||||
* connection must still be subject to exponential backoff. Before the fix,
|
||||
* onOpen reset the backoff delay to 1s on every cycle, producing a reconnect
|
||||
* attempt every ~3s indefinitely (733 attempts in 40 min for one relay).
|
||||
*/
|
||||
class BasicRelayClientBackoffTest {
|
||||
private val url = NormalizedRelayUrl("wss://flaky.example.com")
|
||||
|
||||
class FakeWebSocket : WebSocket {
|
||||
override fun needsReconnect() = false
|
||||
|
||||
override fun connect() {}
|
||||
|
||||
override fun disconnect() {}
|
||||
|
||||
override fun send(msg: String) = true
|
||||
}
|
||||
|
||||
class FakeWebsocketBuilder : WebsocketBuilder {
|
||||
var connectAttempts = 0
|
||||
lateinit var lastListener: WebSocketListener
|
||||
|
||||
override fun build(
|
||||
url: NormalizedRelayUrl,
|
||||
out: WebSocketListener,
|
||||
): WebSocket {
|
||||
connectAttempts++
|
||||
lastListener = out
|
||||
return FakeWebSocket()
|
||||
}
|
||||
}
|
||||
|
||||
class NoopListener : RelayConnectionListener
|
||||
|
||||
class MutableClock(
|
||||
var now: Long = 1_000_000L,
|
||||
)
|
||||
|
||||
/**
|
||||
* Simulates [totalSeconds] of wall-clock time with a pool tick every
|
||||
* [tickSeconds] (subscription churn calls connectAndSyncFiltersIfDisconnected
|
||||
* roughly this often during active use). [onConnectAttempt] is invoked after
|
||||
* every new socket build so the test can drive the relay's behavior.
|
||||
*/
|
||||
private fun runTicks(
|
||||
client: BasicRelayClient,
|
||||
builder: FakeWebsocketBuilder,
|
||||
clock: MutableClock,
|
||||
totalSeconds: Long,
|
||||
tickSeconds: Long = 3,
|
||||
onConnectAttempt: (WebSocketListener) -> Unit,
|
||||
) {
|
||||
val end = clock.now + totalSeconds
|
||||
while (clock.now < end) {
|
||||
clock.now += tickSeconds
|
||||
val before = builder.connectAttempts
|
||||
client.connectAndSyncFiltersIfDisconnected()
|
||||
if (builder.connectAttempts > before) {
|
||||
onConnectAttempt(builder.lastListener)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun newClient(
|
||||
builder: FakeWebsocketBuilder,
|
||||
clock: MutableClock,
|
||||
) = BasicRelayClient(
|
||||
url = url,
|
||||
socketBuilder = builder,
|
||||
listener = NoopListener(),
|
||||
nowInSeconds = { clock.now },
|
||||
)
|
||||
|
||||
@Test
|
||||
fun handshakeThenResetRelayStillBacksOffExponentially() {
|
||||
val builder = FakeWebsocketBuilder()
|
||||
val clock = MutableClock()
|
||||
val client = newClient(builder, clock)
|
||||
|
||||
// first connection: handshake completes, then the server resets.
|
||||
client.connect()
|
||||
builder.lastListener.onOpen(50, false)
|
||||
builder.lastListener.onFailure(RuntimeException("Connection reset"), null, null)
|
||||
|
||||
// 40 minutes of pool ticks every 3s; the relay flaps on every attempt.
|
||||
runTicks(client, builder, clock, totalSeconds = 40 * 60) { listener ->
|
||||
listener.onOpen(50, false)
|
||||
listener.onFailure(RuntimeException(), null, null)
|
||||
}
|
||||
|
||||
// exponential backoff (1,2,4,...,300s cap) allows at most ~17 attempts
|
||||
// in 40 min. The bug produced one attempt per ~2 ticks (hundreds).
|
||||
assertTrue(
|
||||
builder.connectAttempts <= 20,
|
||||
"Expected exponential backoff to cap reconnects, got ${builder.connectAttempts} attempts in 40 min",
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun preHandshakeFailuresBackOffExponentially() {
|
||||
val builder = FakeWebsocketBuilder()
|
||||
val clock = MutableClock()
|
||||
val client = newClient(builder, clock)
|
||||
|
||||
client.connect()
|
||||
builder.lastListener.onFailure(RuntimeException("SSL handshake failed"), null, null)
|
||||
|
||||
runTicks(client, builder, clock, totalSeconds = 40 * 60) { listener ->
|
||||
listener.onFailure(RuntimeException("SSL handshake failed"), null, null)
|
||||
}
|
||||
|
||||
assertTrue(
|
||||
builder.connectAttempts <= 20,
|
||||
"Expected exponential backoff for pre-handshake failures, got ${builder.connectAttempts} attempts",
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun stableConnectionResetsBackoffAfterDisconnect() {
|
||||
val builder = FakeWebsocketBuilder()
|
||||
val clock = MutableClock()
|
||||
val client = newClient(builder, clock)
|
||||
|
||||
// grow the backoff with a few flapping cycles.
|
||||
client.connect()
|
||||
builder.lastListener.onOpen(50, false)
|
||||
builder.lastListener.onFailure(RuntimeException(), null, null)
|
||||
runTicks(client, builder, clock, totalSeconds = 120) { listener ->
|
||||
listener.onOpen(50, false)
|
||||
listener.onFailure(RuntimeException(), null, null)
|
||||
}
|
||||
|
||||
// now the relay recovers: connection stays up for 10 minutes,
|
||||
// then the server closes it cleanly.
|
||||
runTicks(client, builder, clock, totalSeconds = 15 * 60) { listener ->
|
||||
listener.onOpen(50, false)
|
||||
}
|
||||
check(client.isConnected()) { "Test setup: relay should have reconnected and stayed up" }
|
||||
clock.now += 10 * 60
|
||||
builder.lastListener.onClosed(1000, "server restart")
|
||||
|
||||
// after a long stable session, the relay should reconnect quickly
|
||||
// (within a few ticks), not after the accumulated multi-minute delay.
|
||||
val attemptsBefore = builder.connectAttempts
|
||||
runTicks(client, builder, clock, totalSeconds = 15) { listener ->
|
||||
listener.onOpen(50, false)
|
||||
}
|
||||
assertTrue(
|
||||
builder.connectAttempts > attemptsBefore,
|
||||
"Expected backoff to reset after a stable session: no reconnect within 15s of a clean close",
|
||||
)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user