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() + } + } + } +}