refactor(relay): one adapter is one session; drop the lock and the socket identity checks

The adapters compared the socket OkHttp named in each callback against
the one they held, and took a monitor to make the compare-and-null
atomic and to close a window in connect() where a callback could arrive
before the field was assigned. Neither is needed: the relay client builds
a fresh adapter per dial and OkHttp binds exactly one socket to the
listener created in connect(), so anything that reaches that listener is
from this session by construction. The only question a callback has to
ask is whether the session already ended, which is one AtomicBoolean
claimed by whichever of onClosed, onFailure or disconnect() gets there
first. A compare-and-set keeps the "exactly one terminal report"
guarantee without a monitor, and there is no assignment window left to
guard.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014cq6vrfQkASwxgqXXY4py8

amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/NotificationRelayService.kt

@@ -299,8 +299,7 @@ class NotificationRelayService : Service() {
      *    Tor, network changes). Without this, the client disconnects 30s after
      *    the UI stops collecting.
      *
-     * 2. connectedRelaysFlow: re-read the pool's live relay count (client.connectedRelays())
-     *    and refresh the persistent notification.
+     * 2. connectedRelaysFlow: Updates the persistent notification with relay count.
      *
      * The service does NOT create its own relay subscriptions. Instead, it relies on
      * the AccountFilterAssembler subscription that lives in the Compose tree (LoggedInPage).
@@ -321,31 +320,25 @@ class NotificationRelayService : Service() {
                 }
 
                 launch {
-                    // The flow is only the *trigger*; the number comes from
-                    // client.connectedRelays(), which reads each pool member's live socket
-                    // state. The flow's own value used to over-report: it is fed by socket
-                    // callbacks, and until the OkHttp adapters answered a relay's CLOSE frame
-                    // a relay-initiated close produced none (no onClosed, no onFailure, and a
-                    // silent cancel() afterwards). After the feeds tore down in the background
-                    // that left hundreds of already-dropped relays in it for minutes, with no
-                    // subscription to justify a single one of them. The pool now clears the
-                    // flow itself when it lets a relay go and every transport reports its
-                    // session end exactly once, so the flow moves on every change that matters
-                    // and is a sufficient trigger; the members are still read directly because
-                    // that is the ground truth the count is meant to show.
+                    // This flow used to over-report: it is fed by socket callbacks, and until
+                    // the OkHttp adapters answered a relay's CLOSE frame a relay-initiated close
+                    // produced none (no onClosed, no onFailure, and a silent cancel()
+                    // afterwards), so after the feeds tore down in the background it carried
+                    // hundreds of already-dropped relays for minutes. The pool now clears it
+                    // itself whenever it lets a relay go, and every transport reports its
+                    // session end exactly once (see WebSocket), so what it emits is the count.
                     //
                     // sample() caps how often we touch the notification. During feed
-                    // load/teardown these flows churn dozens of times per second; posting on
-                    // every delta blows past Android's notification rate limit (~10/s), which
-                    // silently drops updates and leaves the visible count stuck on a stale
-                    // intermediate value. One refresh per second stays well under the limit
-                    // and always lands the settled count.
-                    val client = Amethyst.instance.client
-                    client
+                    // load/teardown connectedRelaysFlow churns dozens of times per second;
+                    // posting on every delta blows past Android's notification rate limit
+                    // (~10/s), which silently drops updates and leaves the visible count
+                    // stuck on a stale intermediate value. One refresh per second stays
+                    // well under the limit and always lands the settled count.
+                    Amethyst.instance.client
                         .connectedRelaysFlow()
                         .sample(NOTIFICATION_REFRESH_MS)
-                        .collectLatest {
-                            val count = client.connectedRelays().size
+                        .collectLatest { relays ->
+                            val count = relays.size
                             if (count != connectedRelayCount) {
                                 connectedRelayCount = count
                                 updateNotification(count)

amethyst/src/main/java/com/vitorpamplona/amethyst/service/notifications/RelayPurposeSummary.kt

@@ -56,10 +56,7 @@ object RelayPurposeSummary {
         val named = mutableMapOf<SubPurpose, MutableSet<NormalizedRelayUrl>>()
         val browsing = mutableSetOf<NormalizedRelayUrl>()
 
-        // Same source as the count above it (see NotificationRelayService): the pool's live socket
-        // state, not the callback-maintained flow, so the breakdown never has to explain relays the
-        // pool has already dropped.
-        client.connectedRelays().forEach { relay ->
+        client.connectedRelaysFlow().value.forEach { relay ->
             client
                 .activeRequests(relay)
                 .values

amethyst/src/main/java/com/vitorpamplona/amethyst/service/okhttp/OkHttpWebSocket.kt

@@ -34,24 +34,26 @@ import kotlinx.coroutines.launch
 import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.Response
+import java.util.concurrent.atomic.AtomicBoolean
 
 class OkHttpWebSocket(
     val url: NormalizedRelayUrl,
     val httpClient: (url: NormalizedRelayUrl) -> OkHttpClient,
     val out: WebSocketListener,
 ) : WebSocket {
-    private val lock = Any()
     private var usingOkHttp: OkHttpClient? = null
 
-    /**
-     * The OkHttp socket this adapter currently owns, or null once the session has ended -- by the
-     * relay closing it, by a network failure, or by [disconnect]. Only the owned socket may reach
-     * [out], and the terminal callbacks claim the slot under [lock], so a session ends with exactly
-     * one report however it ends. See quartz's `BasicOkHttpWebSocket` for the full reasoning; the
-     * two adapters differ only in how [needsReconnect] is decided.
-     */
     @Volatile private var socket: okhttp3.WebSocket? = null
 
+    /**
+     * Set once, by whichever of `onClosed`, `onFailure` or [disconnect] ends the session first.
+     * One adapter is one session (the relay client builds a fresh one per dial, and OkHttp binds
+     * exactly one socket to the listener), so a callback only has to ask whether the session
+     * already ended. See quartz's `BasicOkHttpWebSocket` for the full reasoning; the two adapters
+     * differ only in how [needsReconnect] is decided.
+     */
+    private val ended = AtomicBoolean(false)
+
     fun buildRequest() = Request.Builder().url(url.url).build()
 
     override fun needsReconnect(): Boolean {
@@ -77,13 +79,10 @@ class OkHttpWebSocket(
     }
 
     override fun connect() {
+        if (socket != null || ended.get()) return
         val client = httpClient(url)
-        // Under the lock so a callback racing this dial waits until the socket is owned rather
-        // than being dropped as foreign.
-        synchronized(lock) {
-            usingOkHttp = client
-            socket = client.newWebSocket(buildRequest(), OkHttpWebsocketListener(out))
-        }
+        usingOkHttp = client
+        socket = client.newWebSocket(buildRequest(), OkHttpWebsocketListener(out))
     }
 
     inner class OkHttpWebsocketListener(
@@ -105,25 +104,21 @@ class OkHttpWebSocket(
                 }
             }
 
-        /** Only the socket this adapter still owns may reach [out]. */
-        private fun isOwned(webSocket: okhttp3.WebSocket) = synchronized(lock) { socket === webSocket }
-
         /** Claims the session's single terminal report. False if it already ended. */
-        private fun endSession(webSocket: okhttp3.WebSocket): Boolean {
-            val ended = synchronized(lock) { (socket === webSocket).also { if (it) socket = null } }
-            if (ended) {
-                incomingMessages.close()
-                job.cancel()
-                scope.cancel()
-            }
-            return ended
+        private fun endSession(): Boolean {
+            if (!ended.compareAndSet(false, true)) return false
+            socket = null
+            incomingMessages.close()
+            job.cancel()
+            scope.cancel()
+            return true
         }
 
         override fun onOpen(
             webSocket: okhttp3.WebSocket,
             response: Response,
         ) {
-            if (!isOwned(webSocket)) return
+            if (ended.get()) return
             out.onOpen(
                 (response.receivedResponseAtMillis - response.sentRequestAtMillis).toInt(),
                 response.headers["Sec-WebSocket-Extensions"]?.contains("permessage-deflate") ?: false,
@@ -134,7 +129,7 @@ class OkHttpWebSocket(
             webSocket: okhttp3.WebSocket,
             text: String,
         ) {
-            if (!isOwned(webSocket)) return
+            if (ended.get()) return
             // Never blocks (unlimited channel): the OkHttp reader thread must
             // stay free to keep draining the socket.
             incomingMessages.trySendBlocking(text)
@@ -162,7 +157,7 @@ class OkHttpWebSocket(
             code: Int,
             reason: String,
         ) {
-            if (!endSession(webSocket)) return
+            if (!endSession()) return
             out.onClosed(code, reason)
         }
 
@@ -171,7 +166,7 @@ class OkHttpWebSocket(
             t: Throwable,
             response: Response?,
         ) {
-            if (!endSession(webSocket)) return
+            if (!endSession()) return
             out.onFailure(t, response?.code, response?.message)
         }
     }
@@ -195,7 +190,9 @@ class OkHttpWebSocket(
         // waiting): OkHttp's cancel() raises no callback when no reader is left to fail, and when
         // it does the failure arrives later on its own thread. The relay client needs the answer
         // now, and must not hear from this socket again.
-        val closing = synchronized(lock) { socket?.also { socket = null } } ?: return
+        val closing = socket ?: return
+        if (!ended.compareAndSet(false, true)) return
+        socket = null
         closing.cancel()
         out.onClosed(1000, "client disconnect")
     }

amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/relays/subscriptions/ActiveSubscriptionsViewModel.kt

@@ -156,8 +156,7 @@ class ActiveSubscriptionsViewModel : ViewModel() {
         withContext(Dispatchers.Default) {
             val client = Amethyst.instance.client
             aggregateSubscriptions(
-                // The pool's live socket state, same as the always-on notification this screen explains.
-                client.connectedRelays().associateWith { relay ->
+                client.connectedRelaysFlow().value.associateWith { relay ->
                     client.activeRequests(relay).values.flatten()
                 },
             )

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/INostrClient.kt

@@ -37,16 +37,6 @@ interface INostrClient : AutoCloseable {
 
     fun availableRelaysFlow(): StateFlow<Set<NormalizedRelayUrl>>
 
-    /**
-     * The relays whose socket is up at the moment of the call, read from the pool's members rather
-     * than from the callback-maintained [connectedRelaysFlow]. The flow is for reacting to changes;
-     * this is for reporting a count: it is computed from what the pool actually holds, so a terminal
-     * callback the socket layer never delivered cannot leave a relay in it that the pool has already
-     * dropped (see `RelayPool.connectedRelayUrls`). Defaults to the flow's current value for clients
-     * that have no pool behind them.
-     */
-    fun connectedRelays(): Set<NormalizedRelayUrl> = connectedRelaysFlow().value
-
     fun connect()
 
     fun disconnect()

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/NostrClient.kt

@@ -493,8 +493,6 @@ class NostrClient(
 
     override fun connectedRelaysFlow() = relayPool.connectedRelays
 
-    override fun connectedRelays() = relayPool.connectedRelayUrls()
-
     override fun availableRelaysFlow() = relayPool.availableRelays
 
     override fun close() {

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPool.kt

@@ -119,8 +119,9 @@ class RelayPool(
         relays.forEach { url, relay ->
             relay.disconnect()
         }
-        // We just tore every socket down; don't leave the answer to the socket layer's
-        // callbacks (see [connectedRelayUrls] for how those can go missing).
+        // We just tore every socket down; say so here rather than leaving it to each socket's
+        // own report. The transports do report a disconnect synchronously (see [WebSocket]),
+        // but this flow is what the app reads as "connected", and it must not depend on that.
         _connectedRelays.update { emptySet() }
     }
 
@@ -274,24 +275,4 @@ class RelayPool(
     ) = listener.onSent(relay, cmdStr, cmd, success)
 
     fun connectedRelaysCount(): Int = relays.count { url, relay -> relay.isConnected() }
-
-    /**
-     * The relays whose socket is up *right now*, read from each pool member's [IRelayClient.isConnected].
-     *
-     * [connectedRelays] is a projection of this that moves on the [onConnected] / [onDisconnected]
-     * callbacks plus the pool's own removals and [disconnect]. The two can still drift for a relay that
-     * is *still a member* whose socket layer lost a terminal callback: before the OkHttp sockets
-     * answered a relay's CLOSE frame, that was every relay-initiated close (no `onClosed`, no
-     * `onFailure`, and a silent `cancel()` afterwards), and the flow carried such relays for minutes
-     * after the pool had let go of them. Reading the members directly cannot be fooled by a callback
-     * that never came for a relay the pool no longer holds. Use this for anything a person reads as
-     * "how many relays am I connected to"; keep the flow for change notification.
-     */
-    fun connectedRelayUrls(): Set<NormalizedRelayUrl> {
-        val urls = mutableSetOf<NormalizedRelayUrl>()
-        relays.forEach { url, relay ->
-            if (relay.isConnected()) urls.add(url)
-        }
-        return urls
-    }
 }

quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/pool/RelayPoolConnectedFlowTest.kt

@@ -26,59 +26,56 @@ import kotlin.test.Test
 import kotlin.test.assertEquals
 
 /**
- * Both views of "connected" -- the callback-fed [RelayPool.connectedRelays] flow and the
- * members-read [RelayPool.connectedRelayUrls] snapshot -- must agree the moment the pool lets a
- * relay go, even when the socket layer never confirms the close.
+ * [RelayPool.connectedRelays] is what the app reads as "how many relays am I connected to", so it
+ * must drop a relay the moment the pool lets go of it, even when the socket layer never confirms
+ * the close.
  *
  * [com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.FakeWebSocket.disconnect] is a
  * no-op that never calls back, which is exactly what OkHttp did after a relay sent a CLOSE frame
- * the app never answered: `cancel()` then fired neither `onClosed` nor `onFailure`. The flow used
- * to keep such a relay for minutes; now the pool clears it on removal and on disconnect itself
- * (and the real transports report a disconnect synchronously, see [com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket]),
- * so neither view depends on a callback that may never come.
+ * the app never answered: `cancel()` then fired neither `onClosed` nor `onFailure`, and the flow
+ * carried such relays for minutes. The real transports now report a disconnect synchronously (see
+ * [com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocket]), but this flow must not depend on
+ * that: the pool clears it itself on removal and on disconnect.
  */
-class RelayPoolConnectedSnapshotTest {
+class RelayPoolConnectedFlowTest {
     private val url = NormalizedRelayUrl("wss://relay.example.com/")
 
-    private fun openedPool(): Pair<FakeWebsocketBuilder, RelayPool> {
+    private fun openedPool(): RelayPool {
         val sockets = FakeWebsocketBuilder()
         val pool = RelayPool(sockets)
         pool.getOrCreateRelay(url).connect()
         sockets.lastListener.onOpen(pingMillis = 10, compression = false)
-        return sockets to pool
+        return pool
     }
 
     @Test
-    fun snapshotMatchesTheFlowWhileTheSocketIsOpen() {
-        val (_, pool) = openedPool()
+    fun anOpenSocketIsConnected() {
+        val pool = openedPool()
 
         assertEquals(setOf(url), pool.connectedRelays.value)
-        assertEquals(setOf(url), pool.connectedRelayUrls())
         assertEquals(1, pool.connectedRelaysCount())
     }
 
     @Test
-    fun removingARelayClearsBothViewsEvenWhenTheCloseIsSilent() {
-        val (_, pool) = openedPool()
+    fun removingARelayDropsItEvenWhenTheCloseIsSilent() {
+        val pool = openedPool()
 
         // No subscription wants it anymore: the pool drops it and cancels its socket, and the
         // socket layer stays silent.
         pool.removeRelay(url)
 
         assertEquals(emptySet(), pool.connectedRelays.value)
-        assertEquals(emptySet(), pool.connectedRelayUrls())
         assertEquals(0, pool.connectedRelaysCount())
     }
 
     @Test
-    fun disconnectingThePoolClearsBothViewsEvenWhenTheCloseIsSilent() {
-        val (_, pool) = openedPool()
+    fun disconnectingThePoolDropsEveryRelayEvenWhenTheCloseIsSilent() {
+        val pool = openedPool()
 
         // The host is putting the client down (app backgrounded, connectivity lost).
         pool.disconnect()
 
         assertEquals(emptySet(), pool.connectedRelays.value)
-        assertEquals(emptySet(), pool.connectedRelayUrls())
         assertEquals(setOf(url), pool.availableRelays.value, "still a member, just not connected")
     }
 }

quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/sockets/okhttp/BasicOkHttpWebSocket.kt

@@ -35,6 +35,7 @@ import kotlinx.coroutines.launch
 import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.Response
+import java.util.concurrent.atomic.AtomicBoolean
 import okhttp3.WebSocket as OkHttpWebSocket
 import okhttp3.WebSocketListener as OkHttpWebSocketListener
 
@@ -51,25 +52,27 @@ class BasicOkHttpWebSocket(
             }
     }
 
-    private val lock = Any()
+    @Volatile private var socket: OkHttpWebSocket? = null
 
     /**
-     * The OkHttp socket this adapter currently owns, or null once the session has ended -- by the
-     * relay closing it, by a network failure, or by [disconnect].
+     * Set once, by whichever of `onClosed`, `onFailure` or [disconnect] ends the session first.
      *
-     * OkHttp names the socket in every callback, and only the owned one may reach [out]. That is
-     * what makes this adapter honour the [WebSocket.disconnect] contract: after [disconnect] the
-     * slot is empty, so the failure OkHttp raises for its own `cancel()` on the reader thread,
-     * or the `onClosed` its writer thread delivers once a close handshake completes, is dropped
-     * instead of reaching a relay client that has already moved on to a new socket. The terminal
-     * callbacks claim the slot under [lock], so a session ends with exactly one report however it
-     * ends.
+     * One adapter is one session: the relay client builds a fresh one per dial, and OkHttp binds
+     * exactly one socket to the listener created in [connect], so anything that reaches that
+     * listener is from this session by construction. The only question a callback has to ask is
+     * whether the session already ended -- which is what keeps the [WebSocket.disconnect] contract:
+     * after [disconnect] the failure OkHttp raises for its own `cancel()` on the reader thread, or
+     * the `onClosed` its writer thread delivers once a close handshake completes, is dropped rather
+     * than reaching a relay client that has already moved on. Claimed with a compare-and-set so a
+     * [disconnect] racing a terminal callback still yields exactly one report.
      */
-    @Volatile private var socket: OkHttpWebSocket? = null
+    private val ended = AtomicBoolean(false)
 
     override fun needsReconnect() = socket == null
 
     override fun connect() {
+        if (socket != null || ended.get()) return
+
         val request = Request.Builder().url(url.url).build()
 
         val listener =
@@ -93,25 +96,21 @@ class BasicOkHttpWebSocket(
                         }
                     }
 
-                /** Only the socket this adapter still owns may reach [out]. */
-                private fun isOwned(webSocket: OkHttpWebSocket) = synchronized(lock) { socket === webSocket }
-
                 /** Claims the session's single terminal report. False if it already ended. */
-                private fun endSession(webSocket: OkHttpWebSocket): Boolean {
-                    val ended = synchronized(lock) { (socket === webSocket).also { if (it) socket = null } }
-                    if (ended) {
-                        incomingMessages.close()
-                        job.cancel()
-                        scope.cancel()
-                    }
-                    return ended
+                private fun endSession(): Boolean {
+                    if (!ended.compareAndSet(false, true)) return false
+                    socket = null
+                    incomingMessages.close()
+                    job.cancel()
+                    scope.cancel()
+                    return true
                 }
 
                 override fun onOpen(
                     webSocket: OkHttpWebSocket,
                     response: Response,
                 ) {
-                    if (!isOwned(webSocket)) return
+                    if (ended.get()) return
                     out.onOpen(
                         (response.receivedResponseAtMillis - response.sentRequestAtMillis).toInt(),
                         response.headers["Sec-WebSocket-Extensions"]?.contains("permessage-deflate") ?: false,
@@ -122,7 +121,7 @@ class BasicOkHttpWebSocket(
                     webSocket: OkHttpWebSocket,
                     text: String,
                 ) {
-                    if (!isOwned(webSocket)) return
+                    if (ended.get()) return
                     // Never blocks (unlimited channel): the OkHttp reader
                     // thread must stay free to keep draining the socket.
                     incomingMessages.trySendBlocking(text)
@@ -155,7 +154,7 @@ class BasicOkHttpWebSocket(
                     code: Int,
                     reason: String,
                 ) {
-                    if (!endSession(webSocket)) return
+                    if (!endSession()) return
                     out.onClosed(code, reason)
                 }
 
@@ -164,23 +163,21 @@ class BasicOkHttpWebSocket(
                     t: Throwable,
                     response: Response?,
                 ) {
-                    if (!endSession(webSocket)) return
+                    if (!endSession()) return
                     out.onFailure(t, response?.code, response?.message)
                 }
             }
 
-        // Under the lock so a callback racing this dial (an instant failure lands on another
-        // thread) waits until the socket is owned, rather than being dropped as foreign.
-        synchronized(lock) {
-            socket = httpClient(url).newWebSocket(request, listener)
-        }
+        socket = httpClient(url).newWebSocket(request, listener)
     }
 
     override fun disconnect() {
         // Claim the session ourselves: OkHttp's cancel() raises no callback when no reader is
         // left to fail (the state a relay-initiated close leaves behind), and when it does the
         // failure arrives later on its own thread. The relay client needs the answer now.
-        val closing = synchronized(lock) { socket?.also { socket = null } } ?: return
+        val closing = socket ?: return
+        if (!ended.compareAndSet(false, true)) return
+        socket = null
         closing.cancel()
         out.onClosed(1000, "client disconnect")
     }