Merge pull request #3838 from vitorpamplona/claude/nip42-auth-dm-delivery-test-fhn7fd

Fix InProcessWebSocket race condition in connect-time AUTH delivery
2026-08-09 08:04:45 +00:00 · 2026-08-01 01:29:24 -04:00
parent 79f198c729 ae61e69136
commit 2741d32cfd
2 changed files with 207 additions and 4 deletions
@@ -41,7 +41,19 @@ import kotlinx.coroutines.launch
 *  - Outbound (`send`) → server `RelaySession.receive()` via an inbound
 *    channel drained by a single coroutine, preserving message order
 *    per the [WebSocketListener] contract.
- *  - Server-side `send` callbacks → [WebSocketListener.onMessage].
+ *  - Server-side `send` callbacks → [WebSocketListener.onMessage], via an
+ *    outbound channel drained by a single coroutine started only after
+ *    [WebSocketListener.onOpen] has fired.
+ *
+ * The outbound channel is not an optimization: a session's connect-time
+ * policies send synchronously from inside `server.connect` (e.g.
+ * [com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy]'s
+ * AUTH challenge), which is before this socket has stored its own state and
+ * before `onOpen`. Delivering those frames directly would break the
+ * [WebSocketListener] contract (no `onMessage` before `onOpen`) and — worse —
+ * a listener that answers the challenge from another thread (RelayAuthenticator
+ * signs and replies concurrently) could hit [send] while `incoming` is still
+ * null, silently losing the reply and deadlocking the NIP-42 handshake.
 *
 * Use this to wire a `NostrClient` to an embedded server in unit tests
 * or single-JVM scenarios without paying for a real TCP socket. Because
@@ -49,8 +61,8 @@ import kotlinx.coroutines.launch
 * expects.
 *
 * Reconnect-after-disconnect is supported: each [connect] creates a
- * fresh scope + drain channel so a previous [disconnect] (which
- * cancels both) doesn't leave a dead drainer behind.
+ * fresh scope + drain channels so a previous [disconnect] (which
+ * cancels them) doesn't leave a dead drainer behind.
 */
 class InProcessWebSocket(
    private val server: NostrServer,
@@ -58,7 +70,9 @@ class InProcessWebSocket(
 ) : WebSocket {
    private var scope: CoroutineScope? = null
    private var incoming: Channel<String>? = null
+    private var outgoing: Channel<String>? = null
    private var drainJob: Job? = null
+    private var deliverJob: Job? = null
    private var session: RelaySession? = null

    override fun needsReconnect(): Boolean = session == null
@@ -67,10 +81,12 @@ class InProcessWebSocket(
        if (session != null) return
        val newScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
        val newIncoming = Channel<String>(UNLIMITED)
-        val s = server.connect { json -> out.onMessage(json) }
+        val newOutgoing = Channel<String>(UNLIMITED)
+        val s = server.connect { json -> newOutgoing.trySend(json) }

        scope = newScope
        incoming = newIncoming
+        outgoing = newOutgoing
        session = s
        drainJob =
            newScope.launch {
@@ -80,6 +96,15 @@ class InProcessWebSocket(
            }

        out.onOpen(0, false)
+
+        // Started only after onOpen so every buffered connect-time frame (AUTH
+        // challenge & co.) reaches the listener with the socket fully wired.
+        deliverJob =
+            newScope.launch {
+                for (msg in newOutgoing) {
+                    out.onMessage(msg)
+                }
+            }
    }

    override fun disconnect() {
@@ -87,7 +112,10 @@ class InProcessWebSocket(
        session = null
        incoming?.close()
        incoming = null
+        outgoing?.close()
+        outgoing = null
        drainJob = null
+        deliverJob = null
        scope?.cancel()
        scope = null
        s.close()
@@ -0,0 +1,175 @@
+/*
+ * 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.server.inprocess
+
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip01Core.relay.server.NostrServer
+import com.vitorpamplona.quartz.nip01Core.relay.server.policies.FullAuthPolicy
+import com.vitorpamplona.quartz.nip01Core.relay.sockets.WebSocketListener
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * Pins the [WebSocketListener] contract on the in-process transport against a
+ * server whose policy sends from inside `onConnect` — [FullAuthPolicy] pushes
+ * its AUTH challenge synchronously while `server.connect` is still on the
+ * stack, before the socket has stored its own state.
+ *
+ * Both tests reproduce (pre-fix, deterministically) the CI-only stall in
+ * geode's Nip42AuthDmDeliveryTest: the challenge used to be delivered before
+ * `onOpen` and before `incoming` was assigned, so a listener that answered it
+ * concurrently (RelayAuthenticator signs on its own coroutine) could call
+ * [InProcessWebSocket.send] on a half-built socket, get `false`, and lose the
+ * AUTH reply forever — the challenge is dedup'd as already-answered, an EVENT
+ * rejected `auth-required:` never re-triggers auth, and the DM never lands.
+ */
+class InProcessWebSocketTest {
+    private val relayUrl = NormalizedRelayUrl("wss://relay.example.com/")
+
+    private fun newServer() =
+        NostrServer(
+            store = EventStore(null),
+            policyBuilder = { FullAuthPolicy(relayUrl) },
+        )
+
+    @Test
+    fun onOpenPrecedesEveryMessage() =
+        runTest {
+            withContext(Dispatchers.Default) {
+                val server = newServer()
+                val callbacks = Channel<String>(UNLIMITED)
+
+                val listener =
+                    object : WebSocketListener {
+                        override fun onOpen(
+                            pingMillis: Int,
+                            compression: Boolean,
+                        ) {
+                            callbacks.trySend("open")
+                        }
+
+                        override fun onMessage(text: String) {
+                            callbacks.trySend("message")
+                        }
+
+                        override fun onClosed(
+                            code: Int,
+                            reason: String,
+                        ) {
+                        }
+
+                        override fun onFailure(
+                            t: Throwable,
+                            code: Int?,
+                            response: String?,
+                        ) {
+                        }
+                    }
+
+                val socket = InProcessWebSocket(server, listener)
+                try {
+                    socket.connect()
+
+                    // FullAuthPolicy sends its AUTH challenge at connect time, so both
+                    // callbacks are guaranteed to arrive; the contract is their order.
+                    val received = mutableListOf<String>()
+                    withTimeout(5_000) {
+                        while ("message" !in received) received.add(callbacks.receive())
+                    }
+
+                    assertEquals(
+                        "open",
+                        received.first(),
+                        "the connect-time AUTH challenge must not be delivered before onOpen; got $received",
+                    )
+                } finally {
+                    socket.disconnect()
+                    server.close()
+                }
+            }
+        }
+
+    @Test
+    fun replySentFromChallengeHandlerIsNotDropped() =
+        runTest {
+            withContext(Dispatchers.Default) {
+                val server = newServer()
+
+                var socket: InProcessWebSocket? = null
+                val replyAccepted = Channel<Boolean>(UNLIMITED)
+
+                val listener =
+                    object : WebSocketListener {
+                        override fun onOpen(
+                            pingMillis: Int,
+                            compression: Boolean,
+                        ) {
+                        }
+
+                        override fun onMessage(text: String) {
+                            // Answer the AUTH challenge immediately, the way
+                            // RelayAuthenticator does. The socket must be fully
+                            // wired by the time any server frame is delivered,
+                            // so this send must be accepted — a `false` here is
+                            // a silently lost AUTH and a dead NIP-42 handshake.
+                            replyAccepted.trySend(socket?.send("""["CLOSE","probe"]""") == true)
+                        }
+
+                        override fun onClosed(
+                            code: Int,
+                            reason: String,
+                        ) {
+                        }
+
+                        override fun onFailure(
+                            t: Throwable,
+                            code: Int?,
+                            response: String?,
+                        ) {
+                        }
+                    }
+
+                val s = InProcessWebSocket(server, listener)
+                socket = s
+                try {
+                    s.connect()
+
+                    val accepted = withTimeout(5_000) { replyAccepted.receive() }
+
+                    assertTrue(
+                        accepted,
+                        "a reply sent from the first onMessage must reach the server, not be dropped",
+                    )
+                } finally {
+                    s.disconnect()
+                    server.close()
+                }
+            }
+        }
+}