From ae61e69136fd64a8bdbb18b9d83b57a8aec1fd07 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 1 Aug 2026 05:27:51 +0000 Subject: [PATCH] fix(relays): deliver in-process server frames only after onOpen MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nip42AuthDmDeliveryTest stalled for its full 10s timeout on CI while passing locally. The stall is a race in InProcessWebSocket.connect(): server.connect() runs the session's connect-time policies synchronously, so FullAuthPolicy's AUTH challenge reached the client's listener before the socket assigned its `incoming` channel and before onOpen fired — breaking the WebSocketListener contract (no onMessage before onOpen). RelayAuthenticator answers that challenge on its own coroutine. When the signed AUTH reply hit send() before the connect thread reached the `incoming` assignment, send() returned false and the reply was silently dropped. Nothing recovers from that: the challenge is already dedup'd as answered, and an EVENT rejected with OK-false `auth-required:` never re-triggers auth (only a CLOSED does), so the pending gift wrap was never resent — exactly the CI signature (10.011s, no auth activity between the authenticator's Init and Destroy logs). Server->client frames now go through an outbound channel drained by a coroutine started only after onOpen, so every connect-time frame reaches the listener with the socket fully wired. Order is preserved by the single drainer, same as the existing inbound path. Both new InProcessWebSocketTest cases fail deterministically without the reorder (the challenge always outran onOpen; a reply sent from the first onMessage was always rejected) and pass with it, on top of the full :geode:test and :quartz:jvmTest suites. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Bf1Y91sfjxwi2ig4ymGTA9 --- .../server/inprocess/InProcessWebSocket.kt | 36 +++- .../inprocess/InProcessWebSocketTest.kt | 175 ++++++++++++++++++ 2 files changed, 207 insertions(+), 4 deletions(-) create mode 100644 quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/inprocess/InProcessWebSocketTest.kt diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/inprocess/InProcessWebSocket.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/inprocess/InProcessWebSocket.kt index 69def1b78c..0b3672e445 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/inprocess/InProcessWebSocket.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/inprocess/InProcessWebSocket.kt @@ -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? = null + private var outgoing: Channel? = 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(UNLIMITED) - val s = server.connect { json -> out.onMessage(json) } + val newOutgoing = Channel(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() diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/inprocess/InProcessWebSocketTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/inprocess/InProcessWebSocketTest.kt new file mode 100644 index 0000000000..2b9cda14cd --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/server/inprocess/InProcessWebSocketTest.kt @@ -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(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() + 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(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() + } + } + } +}