feat(relays): reconnect dropped sockets and replay what they were carrying

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 <noreply@anthropic.com>

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<String, MutableMap<String, String>>()
+
+    private val retainedRequestsMutex = Mutex()
+
     @VisibleForTesting
     var socketClients = setOf<press.mantra.compose.network.sockets.NostrSocketClient>()
 
@@ -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<press.mantra.compose.network.dto.RelayDTO>) {
         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<String>) {
+        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<press.mantra.compose.network.dto.RelayDTO>.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)
 

composeApp/src/commonMain/kotlin/press/mantra/compose/network/sockets/NostrSocketClient.kt

@@ -9,6 +9,15 @@ interface NostrSocketClient {
 
     val incomingMessages: SharedFlow<NostrIncomingMessage>
 
+    /**
+     * 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(

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,
     )
 }

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<NostrIncomingMessage>()
     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) {

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