From c3b0c8671bfbe217e322a760398cefd84517b7bd Mon Sep 17 00:00:00 2001 From: Kgothatso Ngako Date: Sat, 5 Sep 2026 16:02:11 +0200 Subject: [PATCH] feat(relays): reconnect dropped sockets and replay what they were carrying MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing in this app reconnected a websocket. NostrSocketClientImpl caught a failure, called close(), and fired onSocketConnectionClosed — which only flips a boolean in relayPoolStatus that nobody reads. A socket that failed stayed failed, and the only reason that was survivable is that every subscription is short: the next queued request opens a fresh socket on its way out through ensureSocketConnectionOrThrow. That stops being survivable the moment a subscription is meant to outlive the socket, so this lands first, on its own. It is already a fix without any of that: a REQ interrupted mid-download used to sit there until SUBSCRIPTION_TIMEOUT gave up 120s later, having saved whatever partial set arrived before the drop. Now the socket comes back and the REQ is re-sent. The socket client: - a supervised reconnect loop with exponential backoff (1s doubling to 60s) plus up to 25% jitter, because every relay in the pool drops at once when the network does and without jitter they all come back in lockstep. The exponent is capped so a socket failing for hours cannot overflow the doubling into Infinity, which Duration * Double rejects outright. - `autoReconnect`, off by default and owned by the pool. Reconnecting a socket nobody is subscribed on is battery spent on nothing, so the pool turns it on for exactly as long as it retains a subscription for that relay. - `closedByClient`, so closePool() is not answered by every socket in it politely reconnecting. Cleared by the next caller-driven connect. - onSessionLost() as the single exit point for a session that ended without the client asking, replacing the close()-from-inside-the-receiver dance. It identity-checks the session before clearing it, so a reconnect that already installed a newer one is not torn down by its predecessor's cleanup, and runs NonCancellable because the receiver job is cancelled as part of a replacement connect. - Frame.Close now breaks the receive loop rather than closing by hand. The relay closing us is not the client closing us, so it earns a reconnect too. - a new SocketConnectionReopenedCallback, fired only when a session is established on a socket that had connected before. Kept separate from "opened" because on a FIRST connect a replay would double-send the very REQ whose sendMESSAGE opened the socket. Two bugs fixed in passing, both of the silent kind: - the compression REQ in the post-connect handshake was written to `wsSession` before the new session was assigned to it, so it went to the previous (usually null) session and was dropped. wsSession is now assigned first. - sendMESSAGE used `wsSession?.send(...)`, so a send on a dropped socket was a no-op and the caller waited forever for an answer to a message never sent. It now warns. The pool: - retains the REQ text per (relay, subscription id), and replays it when that relay's socket is re-established. A relay answers a repeated REQ on the same subscription id by replacing the filter, so replay is a send rather than a close-and-reopen, and the collector already attached to the socket's message flow simply starts receiving again. - retains on query() BEFORE the send, so a socket that dies between there and the relay's first answer is still covered; releases on closeQuery(), which every pump already calls from a NonCancellable finally. - deliberately does NOT retain negentropy. NEG-OPEN carries a fingerprint of the local set and each round depends on the last, so replaying one mid-exchange would reconcile against a conversation the relay is no longer having. An interrupted negentropy request is abandoned and re-queued. - drops retained work for relays removed by changeRelays/removeRelays/ closePool, so a relay edit does not leave a socket reconnecting for subscriptions nobody wants. - collapses the five hand-rolled `socketClients.find { normalize... }` lookups into socketClientFor(), now that there were about to be several more. Co-Authored-By: Claude Opus 5 --- .../compose/network/relays/RelayPool.kt | 119 +++++++++- .../network/sockets/NostrSocketClient.kt | 9 + .../sockets/NostrSocketClientFactory.kt | 4 + .../network/sockets/NostrSocketClientImpl.kt | 210 ++++++++++++++++-- .../sockets/SocketConnectionCallbacks.kt | 10 + 5 files changed, 325 insertions(+), 27 deletions(-) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt index 0d0b0ad0..16b1d44b 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/relays/RelayPool.kt @@ -55,6 +55,20 @@ class RelayPool( private val relayMutex = Mutex() + /** + * The REQ text of every subscription still believed to be open, keyed by normalized + * relay url and then by subscription id, so a socket that comes back can be handed + * its subscriptions again. + * + * Only plain REQs live here. A negentropy exchange is stateful — NEG-OPEN carries a + * fingerprint of the local set and each round depends on the last — so replaying one + * mid-exchange would reconcile against a conversation the relay is no longer having. + * An interrupted negentropy request is abandoned and re-queued instead. + */ + private val retainedRequests = mutableMapOf>() + + private val retainedRequestsMutex = Mutex() + @VisibleForTesting var socketClients = setOf() @@ -75,6 +89,10 @@ class RelayPool( updateRelayStatus(url = url, connected = false) } + private val onSocketConnectionReopenedCallback: press.mantra.compose.network.sockets.SocketConnectionReopenedCallback = { url -> + scope.launch { replayRetainedRequests(url) } + } + fun changeRelays(relays: List) { val existingRelayUrls = socketClients.map { it.socketUrl } val newRelayUrls = relays.map { it.url } @@ -93,6 +111,7 @@ class RelayPool( updateRelayStatus(url = client.socketUrl, connected = false) scope.launch { client.close() } } + forgetRetainedRequests(toRemoveSocketClients.map { it.socketUrl }) this.relays.clear() this.relays.addAll(relays) } @@ -111,6 +130,7 @@ class RelayPool( updateRelayStatus(url = client.socketUrl, connected = false) scope.launch { client.close() } } + forgetRetainedRequests(toRemoveSocketClients.map { it.socketUrl }) this.relays.removeAll(relays) } @@ -139,6 +159,7 @@ class RelayPool( updateRelayStatus(url = client.socketUrl, connected = false) scope.launch { client.close() } } + forgetRetainedRequests(socketClients.map { it.socketUrl }) socketClients = emptySet() relays.clear() } @@ -151,12 +172,92 @@ class RelayPool( } } + private fun socketClientFor(relayUrl: String): press.mantra.compose.network.sockets.NostrSocketClient? { + val wanted = NormalizedRelayUrl(relayUrl).displayUrl() + + return socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == wanted } + } + + /** + * Remembers a subscription so it can be re-sent if this relay's socket comes back, + * and tells the socket it is now worth reconnecting. + */ + private suspend fun retainRequest(relayUrl: String, subId: String, message: String) { + val key = NormalizedRelayUrl(relayUrl).displayUrl() + + retainedRequestsMutex.withLock { + retainedRequests.getOrPut(key) { mutableMapOf() }[subId] = message + } + socketClientFor(relayUrl)?.autoReconnect = true + } + + /** + * Forgets a subscription that has been closed. Once a relay is carrying none of ours, + * its socket stops reconnecting on its own: keeping a socket alive for a relay we have + * nothing open on is battery spent on nothing. + */ + private suspend fun releaseRequest(relayUrl: String, subId: String) { + val key = NormalizedRelayUrl(relayUrl).displayUrl() + + val stillWanted = retainedRequestsMutex.withLock { + val forRelay = retainedRequests[key] + forRelay?.remove(subId) + if (forRelay.isNullOrEmpty()) { + retainedRequests.remove(key) + false + } else { + true + } + } + + if (!stillWanted) socketClientFor(relayUrl)?.autoReconnect = false + } + + private fun forgetRetainedRequests(relayUrls: List) { + if (relayUrls.isEmpty()) return + + val keys = relayUrls.map { NormalizedRelayUrl(it).displayUrl() } + scope.launch { + retainedRequestsMutex.withLock { + keys.forEach { retainedRequests.remove(it) } + } + } + } + + /** + * Hands a relay back the subscriptions it was serving before its socket dropped. + * + * Only ever called for a RE-connection. A relay answers a repeated REQ on the same + * subscription id by replacing the filter, so this is a send rather than a + * close-and-reopen, and the collector already attached to the socket's message flow + * simply starts receiving again. + */ + private suspend fun replayRetainedRequests(relayUrl: String) { + val key = NormalizedRelayUrl(relayUrl).displayUrl() + val messages = retainedRequestsMutex.withLock { retainedRequests[key]?.toMap() } + + if (messages.isNullOrEmpty()) return + + val nostrSocketClient = socketClientFor(relayUrl) + if (nostrSocketClient == null) { + logger.w("Cannot replay ${messages.size} subscription(s): $relayUrl has no socket") + return + } + + logger.i("Replaying ${messages.size} subscription(s) on $relayUrl: ${messages.keys}") + messages.forEach { (subId, message) -> + runCatching { nostrSocketClient.sendMESSAGE(message) } + .onFailure { logger.w(throwable = it) { "Failed to replay $subId on $relayUrl" } } + } + } + private fun List.mapAsNostrSocketClient() = this.map { nostrSocketClientFactory.create( wssUrl = it.url, onSocketConnectionOpened = onSocketConnectionOpenedCallback, onSocketConnectionClosed = onSocketConnectionClosedCallback, + onSocketConnectionReopened = onSocketConnectionReopenedCallback, ) } @@ -186,13 +287,19 @@ class RelayPool( ) logger.d("socketClients: ${socketClients.map { it.socketUrl }}") - val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() } + val nostrSocketClient = socketClientFor(relayUrl) val filterRequest = OptimizedJsonMapper.toJson(reqCommand) if (nostrSocketClient == null) { throw press.mantra.compose.exceptions.NetworkException("$relayUrl is not connected") } + + // Retained BEFORE the send, so a socket that dies between here and the relay's + // first answer still gets the REQ replayed when it comes back. Released by + // closeQuery, which every pump calls from a NonCancellable finally. + retainRequest(relayUrl = relayUrl, subId = reqCommand.subId, message = filterRequest) + return coroutineScope { val eventFlow = nostrSocketClient.queryAsFlow(reqCommand.subId) with(nostrSocketClient) { @@ -210,7 +317,7 @@ class RelayPool( ) logger.d("socketClients: ${socketClients.map { it.socketUrl }}") - val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() } + val nostrSocketClient = socketClientFor(relayUrl) val negentropySyncRequest = OptimizedJsonMapper.toJson(negOpenCmd) @@ -232,20 +339,22 @@ class RelayPool( * [negentropySync] rather than opening a new one. */ suspend fun sendNegentropyMessage(negMsgCmd: NegMsgCmd, relayUrl: String) { - val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() } + val nostrSocketClient = socketClientFor(relayUrl) ?: throw press.mantra.compose.exceptions.NetworkException("$relayUrl is not connected") nostrSocketClient.sendMESSAGE(OptimizedJsonMapper.toJson(negMsgCmd)) } suspend fun closeQuery(closeCmd: CloseCmd, relayUrl: String) { + releaseRequest(relayUrl = relayUrl, subId = closeCmd.subId) + addRelaysIfMissing( setOf( NormalizedRelayUrl(relayUrl).url.toRelayDTO() ) ) - val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() } + val nostrSocketClient = socketClientFor(relayUrl) val closeSubscription = OptimizedJsonMapper.toJson(closeCmd) @@ -267,7 +376,7 @@ class RelayPool( ) logger.d("socketClients: ${socketClients.map { it.socketUrl }}") - val nostrSocketClient = socketClients.find { NormalizedRelayUrl(it.socketUrl).displayUrl() == NormalizedRelayUrl(relayUrl).displayUrl() } + val nostrSocketClient = socketClientFor(relayUrl) val closeNegentropySubscription = OptimizedJsonMapper.toJson(negCloseCmd) diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClient.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClient.kt index 222a6a04..44105ce7 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClient.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClient.kt @@ -9,6 +9,15 @@ interface NostrSocketClient { val incomingMessages: SharedFlow + /** + * Whether a socket that drops should be re-opened on its own, with backoff. + * + * Off by default, and owned by the pool rather than the socket: reconnecting a + * socket nobody is subscribed on is pure battery cost, so the pool turns this on + * for exactly as long as it is retaining at least one subscription for the relay. + */ + var autoReconnect: Boolean + suspend fun close() @Throws( diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientFactory.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientFactory.kt index 735d543b..d751989f 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientFactory.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientFactory.kt @@ -27,6 +27,7 @@ object NostrSocketClientFactory { incomingCompressionEnabled: Boolean = false, onSocketConnectionOpened: SocketConnectionOpenedCallback? = null, onSocketConnectionClosed: SocketConnectionClosedCallback? = null, + onSocketConnectionReopened: SocketConnectionReopenedCallback? = null, ): NostrSocketClient { return NostrSocketClientImpl( httpClient = httpClient, @@ -34,6 +35,7 @@ object NostrSocketClientFactory { incomingCompressionEnabled = incomingCompressionEnabled, onSocketConnectionOpened = onSocketConnectionOpened, onSocketConnectionClosed = onSocketConnectionClosed, + onSocketConnectionReopened = onSocketConnectionReopened, ) } @@ -42,11 +44,13 @@ object NostrSocketClientFactory { incomingCompressionEnabled: Boolean = false, onSocketConnectionOpened: SocketConnectionOpenedCallback? = null, onSocketConnectionClosed: SocketConnectionClosedCallback? = null, + onSocketConnectionReopened: SocketConnectionReopenedCallback? = null, ) = create( httpClient = defaultSocketsHttpClient, wssUrl = wssUrl, incomingCompressionEnabled = incomingCompressionEnabled, onSocketConnectionOpened = onSocketConnectionOpened, onSocketConnectionClosed = onSocketConnectionClosed, + onSocketConnectionReopened = onSocketConnectionReopened, ) } diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientImpl.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientImpl.kt index e5c8235e..aa4442d8 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientImpl.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClientImpl.kt @@ -13,6 +13,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.IO import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableSharedFlow @@ -21,6 +22,7 @@ import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext import kotlinx.serialization.json.JsonObject import okio.Buffer import okio.GzipSink @@ -28,8 +30,14 @@ import okio.Inflater import okio.InflaterSource import okio.buffer import okio.use +import kotlin.concurrent.Volatile import kotlin.coroutines.cancellation.CancellationException +import kotlin.math.min +import kotlin.math.pow +import kotlin.random.Random +import kotlin.time.Duration import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.Duration.Companion.seconds import kotlin.uuid.ExperimentalUuidApi import kotlin.uuid.Uuid @@ -40,15 +48,60 @@ internal class NostrSocketClientImpl( private val incomingCompressionEnabled: Boolean = false, private val onSocketConnectionOpened: SocketConnectionOpenedCallback? = null, private val onSocketConnectionClosed: SocketConnectionClosedCallback? = null, + private val onSocketConnectionReopened: SocketConnectionReopenedCallback? = null, ) : NostrSocketClient { val logger = Logger.withTag("NostrSocketClientImpl") + companion object { + /** Wait before the first reconnect attempt; doubles per failure up to [MAX_RECONNECT_DELAY]. */ + private val INITIAL_RECONNECT_DELAY = 1.seconds + + private val MAX_RECONNECT_DELAY = 60.seconds + + /** + * Bounds the exponent so a socket that has been failing for hours cannot overflow + * the doubling into `Infinity`, which `Duration * Double` rejects outright. + */ + private const val MAX_RECONNECT_EXPONENT = 16 + + /** + * Up to this fraction of the delay is added at random. Every relay in the pool + * drops at once when the network does, and without jitter they would all come + * back in lockstep for as long as the outage lasts. + */ + private const val RECONNECT_JITTER = 0.25 + } + + /** Outcome of a connect attempt, so the caller can tell a first connect from a repair. */ + private enum class ConnectOutcome { ALREADY_CONNECTED, OPENED, REOPENED } + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) private val wsMutex = Mutex() private var wsSession: WebSocketSession? = null private var wsReceiverJob: Job? = null + private var reconnectJob: Job? = null + + /** Consecutive failed reconnect attempts; reset the moment a session is established. */ + private var reconnectAttempts = 0 + + /** + * True once a session has been established, so the next one is a RE-connection and + * the pool owes it a subscription replay. + */ + private var hasConnected = false + + /** + * Set by [close], cleared by the next caller-driven connect. A socket the client + * deliberately closed must stay closed — otherwise `closePool` would be answered by + * every socket in it politely reconnecting. + */ + @Volatile + private var closedByClient = false + + @Volatile + override var autoReconnect: Boolean = false private val _incomingMessages = MutableSharedFlow() override val incomingMessages = _incomingMessages.asSharedFlow() @@ -56,20 +109,21 @@ internal class NostrSocketClientImpl( override val socketUrl = wssUrl.cleanWebSocketUrl() override suspend fun ensureSocketConnectionOrThrow() { - if (wsSession != null && wsSession?.isActive == true) return + if (isSessionActive()) return - wsMutex.withLock { - if (wsSession == null || wsSession?.isActive == false) { - wsSession = acquireWebSocketSession(socketUrl) - } - } - } + val outcome = wsMutex.withLock { + // Asking for the socket is asking for it to stay up: clear the flag close() + // set so a drop after this point reconnects again. + closedByClient = false - private suspend fun acquireWebSocketSession(url: String): WebSocketSession { - return try { - httpClient.webSocketSession(urlString = url).apply { - launchWebSocketReceiver() - onSocketConnectionOpened?.invoke(url) + if (isSessionActive()) { + ConnectOutcome.ALREADY_CONNECTED + } else { + wsSession = openSession() + // Assigned BEFORE the post-connect handshake below, because sendMESSAGE + // writes to `wsSession` rather than to the session it was handed — the + // compression REQ used to go to the previous (usually null) session and + // was silently dropped. if (incomingCompressionEnabled) { val id = Uuid.generateV4().toHexDashString() sendMESSAGE( @@ -77,11 +131,35 @@ internal class NostrSocketClientImpl( ensureSessionBeforeSend = false, ) } + + val reopened = hasConnected + hasConnected = true + reconnectAttempts = 0 + if (reopened) ConnectOutcome.REOPENED else ConnectOutcome.OPENED } + } + + if (outcome == ConnectOutcome.ALREADY_CONNECTED) return + + // Fired outside the lock. A callback that turns around and sends on this socket + // would otherwise queue behind a mutex the caller still holds. + onSocketConnectionOpened?.invoke(socketUrl) + if (outcome == ConnectOutcome.REOPENED) onSocketConnectionReopened?.invoke(socketUrl) + } + + private fun isSessionActive() = wsSession?.isActive == true + + /** Opens a session and starts its receiver. Call with [wsMutex] held. */ + private suspend fun openSession(): WebSocketSession { + return try { + httpClient.webSocketSession(urlString = socketUrl).apply { launchWebSocketReceiver() } } catch (error: Exception) { - logger.w("NostrSocketClient::acquireWebSocketSession($socketUrl) failed.", error) - close() + logger.w("NostrSocketClient::openSession($socketUrl) failed.", error) + wsReceiverJob?.cancel() + wsReceiverJob = null + wsSession = null onSocketConnectionClosed?.invoke(socketUrl, error) + scheduleReconnect() throw press.mantra.compose.exceptions.NetworkException(cause = error) } } @@ -94,6 +172,7 @@ internal class NostrSocketClientImpl( } private suspend fun WebSocketSession.receiveSocketMessages() { + var failure: Throwable? = null try { for (frame in incoming) { when (frame) { @@ -112,8 +191,9 @@ internal class NostrSocketClientImpl( is Frame.Close -> { val closeReason = frame.readReason() logger.w { "WS $socketUrl closed. [${closeReason?.code}, ${closeReason?.message}]" } - close() - onSocketConnectionClosed?.invoke(socketUrl, null) + // Leave the teardown to onSessionLost below. The relay closing us + // is not the client closing us, so this still earns a reconnect. + break } else -> Unit @@ -124,23 +204,102 @@ internal class NostrSocketClientImpl( throw error } catch (error: Exception) { logger.w("NostrSocketClient::receiveSocketMessages() on $socketUrl failed.", error) - close() + failure = error + } + + onSessionLost(session = this, error = failure) + } + + /** + * The single exit point for a session that ended without the client asking. Clears the + * session, tells the pool, and arms the reconnect. + * + * NonCancellable because the receiver job is cancelled as part of a replacement connect, + * and a cancelled cleanup would leave `wsSession` pointing at a dead socket. + */ + private suspend fun onSessionLost(session: WebSocketSession, error: Throwable?) = + withContext(NonCancellable) { + val wasCurrent = wsMutex.withLock { + // Identity check, not a null check: a reconnect may already have installed a + // newer session, and clearing that one would drop a healthy socket. + if (wsSession !== session) { + false + } else { + wsSession = null + true + } + } + + if (!wasCurrent) return@withContext + + runCatching { + session.close( + reason = CloseReason( + code = CloseReason.Codes.NORMAL, + message = "Session ended.", + ), + ) + } onSocketConnectionClosed?.invoke(socketUrl, error) + scheduleReconnect() + } + + /** + * Re-opens a dropped socket with exponential backoff, for as long as the pool says it + * still wants one. Nothing in this app used to reconnect at all: a socket that failed + * stayed failed, and the only reason that was survivable is that every subscription was + * short and the next queued request opened a fresh socket on its way out. + */ + private fun scheduleReconnect() { + if (!autoReconnect || closedByClient) return + if (reconnectJob?.isActive == true) return + + reconnectJob = scope.launch { + while (isActive && autoReconnect && !closedByClient) { + val attempt = ++reconnectAttempts + val wait = reconnectDelay(attempt) + logger.i { "Reconnecting to $socketUrl in $wait (attempt $attempt)" } + delay(wait) + + if (!autoReconnect || closedByClient) return@launch + + val reconnected = runCatching { ensureSocketConnectionOrThrow() } + if (reconnected.isSuccess) { + logger.i { "Reconnected to $socketUrl after $attempt attempt(s)" } + return@launch + } + } } } + private fun reconnectDelay(attempt: Int): Duration { + val doublings = 2.0.pow(min(attempt - 1, MAX_RECONNECT_EXPONENT)) + val backoff = minOf(INITIAL_RECONNECT_DELAY * doublings, MAX_RECONNECT_DELAY) + + return backoff + backoff * RECONNECT_JITTER * Random.nextDouble() + } + override suspend fun close() { - wsReceiverJob?.cancel() - wsReceiverJob = null + val session = wsMutex.withLock { + closedByClient = true + val current = wsSession + wsSession = null + wsReceiverJob?.cancel() + wsReceiverJob = null + current + } + + reconnectJob?.cancel() + reconnectJob = null + runCatching { - wsSession?.close( + session?.close( reason = CloseReason( code = CloseReason.Codes.NORMAL, message = "Closed by client.", ), ) } - wsSession = null } private fun processIncomingMessage(text: String) { @@ -159,7 +318,14 @@ internal class NostrSocketClientImpl( ensureSocketConnectionOrThrow() } logLargeText(text = text, url = socketUrl, incoming = false) - wsSession?.send(Frame.Text(text = text)) + val session = wsSession + if (session == null) { + // Used to be a silent `?.` no-op, which is how a dropped socket could swallow + // a send and leave the caller waiting on an answer to a message never sent. + logger.w { "Dropping a send to $socketUrl: no session" } + return + } + session.send(Frame.Text(text = text)) } override suspend fun sendREQ(subscriptionId: String, data: JsonObject) { diff --git a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/SocketConnectionCallbacks.kt b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/SocketConnectionCallbacks.kt index 580da97b..81f778a4 100644 --- a/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/SocketConnectionCallbacks.kt +++ b/composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/SocketConnectionCallbacks.kt @@ -2,3 +2,13 @@ package press.mantra.compose.network.sockets typealias SocketConnectionOpenedCallback = (url: String) -> Unit typealias SocketConnectionClosedCallback = (url: String, error: Throwable?) -> Unit + +/** + * Fired when a session is established on a socket that had already been connected + * once before — i.e. after a drop, not on the first connect. + * + * Deliberately separate from [SocketConnectionOpenedCallback]. The pool replays the + * subscriptions a socket was carrying when it comes back, and on a FIRST connect + * that replay would double-send the very REQ whose `sendMESSAGE` opened the socket. + */ +typealias SocketConnectionReopenedCallback = (url: String) -> Unit