diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt index 46be99d3bb..d388bf1a2a 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClient.kt @@ -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() } diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClientBackoffTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClientBackoffTest.kt new file mode 100644 index 0000000000..a4f8e83015 --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/single/basic/BasicRelayClientBackoffTest.kt @@ -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", + ) + } +}