diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt new file mode 100644 index 000000000..9c1ce9d34 --- /dev/null +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt @@ -0,0 +1,254 @@ +/* + * 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.nestsclient.interop + +import com.vitorpamplona.nestsclient.NestsRoomConfig +import com.vitorpamplona.nestsclient.OkHttpNestsClient +import com.vitorpamplona.nestsclient.buildRelayConnectTarget +import com.vitorpamplona.nestsclient.connectNestsListener +import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession +import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal +import com.vitorpamplona.quic.tls.PermissiveCertificateValidator +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.flow.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull +import okhttp3.OkHttpClient +import org.junit.AfterClass +import org.junit.BeforeClass +import org.junit.Test +import kotlin.test.fail + +/** + * Local-harness counterpart of + * [NostrnestsProdAudioTransmissionTest.sustained_per_frame_send_outcomes_two_users]. + * + * Drives the local Docker-backed reference relay (kixelated/moq) so we + * can reproduce the production "82/100" frame-loss cliff without needing + * to talk to nostrnests.com. If the same cliff appears here, the bug is + * in our client code (`:nestsClient` / `:quic`); if it doesn't, the bug + * is specific to the production deployment (relay config / network). + * + * Bypasses [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster] + * and calls [com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle.send] + * directly so we can capture the boolean return per frame, plus any + * exception thrown by `endGroup()`. This is the only way to tell whether + * a missing frame was dropped at the moq-lite layer (`send` returned + * false) or queued and lost downstream (`send` returned true but the + * uni-stream write threw, swallowed by `runCatching` inside the + * production `send` itself). + * + * Skipped by default — set `-DnestsInterop=true` to enable. + */ +class NostrNestsSustainedSendOutcomesInteropTest { + @Test + fun two_users_sustained_send_outcomes_against_local_relay() = + runBlocking { + NostrNestsHarness.assumeNestsInterop() + val harness = harnessOrNull ?: return@runBlocking + val scope = "send-trace" + + val hostSigner = NostrSignerInternal(KeyPair()) + val audienceSigner = NostrSignerInternal(KeyPair()) + val room = + NestsRoomConfig( + authBaseUrl = harness.authBaseUrl, + endpoint = harness.moqEndpoint, + hostPubkey = hostSigner.pubKey, + roomId = "send-trace-${System.currentTimeMillis()}", + ) + + val httpClient = OkHttpNestsClient { OkHttpClient() } + // Self-signed dev cert on the harness's moq-relay; production + // tests use the JDK validator. PermissiveCertificateValidator + // is the same one every other harness test in this package + // uses, so we're not deviating from the established pattern. + val transport = + QuicWebTransportFactory(certificateValidator = PermissiveCertificateValidator()) + + val supervisor = SupervisorJob() + val pumpScope = CoroutineScope(supervisor + Dispatchers.IO) + + InteropDebug.checkpoint( + scope, + "host=${hostSigner.pubKey.take(8)}… audience=${audienceSigner.pubKey.take(8)}… " + + "ns=${room.moqNamespace()} frames=$N_FRAMES cadence=${FRAME_CADENCE_MS}ms", + ) + + // ---- speaker side: build session manually so we can call + // session.publish() directly and capture per-frame send results. + val publishToken = + InteropDebug.stepSuspending(scope, "host: mintToken(publish=true)") { + httpClient.mintToken(room = room, publish = true, signer = hostSigner) + } + val (authority, path) = + buildRelayConnectTarget( + endpoint = room.endpoint, + namespace = room.moqNamespace(), + token = publishToken, + ) + val speakerWt = + InteropDebug.stepSuspending(scope, "host: WebTransport.connect") { + transport.connect(authority = authority, path = path, bearerToken = null) + } + val speakerSession = MoqLiteSession.client(speakerWt, pumpScope) + val publisher = + InteropDebug.stepSuspending(scope, "host: session.publish(broadcastSuffix=hostPub)") { + speakerSession.publish(broadcastSuffix = hostSigner.pubKey) + } + + // ---- listener side: production code path, unchanged. + val listener = + InteropDebug.stepSuspending(scope, "audience: connectNestsListener") { + connectNestsListener( + httpClient = httpClient, + transport = transport, + scope = pumpScope, + room = room, + signer = audienceSigner, + ) + } + InteropDebug.assertListenerReached(scope, "Connected", listener.state.value) + + val subscription = + InteropDebug.stepSuspending(scope, "audience: subscribeSpeaker(host)") { + listener.subscribeSpeaker(hostSigner.pubKey) + } + + data class Arrival( + val wallMs: Long, + val groupId: Long, + ) + val received = java.util.concurrent.CopyOnWriteArrayList() + val sendOutcomes = BooleanArray(N_FRAMES) + val endGroupErrors = arrayOfNulls(N_FRAMES) + val collectStart = System.currentTimeMillis() + val collected = + async(pumpScope.coroutineContext) { + withTimeoutOrNull(RECEIVE_TIMEOUT_MS) { + subscription.objects + .onEach { obj -> + received += Arrival(System.currentTimeMillis() - collectStart, obj.groupId) + }.take(N_FRAMES) + .toList() + } + } + + try { + delay(SUBSCRIBE_SETTLE_MS) + + val payloadPrefix = ByteArray(79) { 0x4F.toByte() } + byteArrayOf(0x00) + val started = System.currentTimeMillis() + for (i in 0 until N_FRAMES) { + val payload = payloadPrefix + byteArrayOf(i.toByte()) + sendOutcomes[i] = publisher.send(payload) + runCatching { publisher.endGroup() } + .onFailure { endGroupErrors[i] = it::class.simpleName + ": " + it.message } + delay(FRAME_CADENCE_MS) + } + val pumpDurationMs = System.currentTimeMillis() - started + + collected.await() + val frames = received.toList() + val receivedGroups = frames.map { it.groupId }.toHashSet() + + val sendCount = sendOutcomes.count { it } + val firstFalseSend = sendOutcomes.indexOfFirst { !it } + val sentButLost = (0 until N_FRAMES).filter { sendOutcomes[it] && it.toLong() !in receivedGroups } + val firstSentButLost = sentButLost.firstOrNull() ?: -1 + + InteropDebug.checkpoint( + scope, + "sendTrue=$sendCount/$N_FRAMES received=${frames.size}/$N_FRAMES " + + "firstFalseSend=$firstFalseSend firstSentButLost=$firstSentButLost " + + "pumpDuration=${pumpDurationMs}ms", + ) + for (i in 0 until N_FRAMES) { + val sentOk = sendOutcomes[i] + val recv = i.toLong() in receivedGroups + val tag = + when { + sentOk && recv -> "ok" + sentOk && !recv -> "SENT-LOST" + !sentOk && !recv -> "send=false" + else -> "ghost?" + } + val err = endGroupErrors[i]?.let { ", endGroup err=$it" } ?: "" + InteropDebug.checkpoint(scope, " i=$i $tag$err") + } + + if (firstFalseSend >= 0) { + fail( + "[$scope] publisher.send returned false starting at frame $firstFalseSend — " + + "speaker's MoqLiteSession dropped the frame at the source. " + + "Likely cause: inboundSubs cleared (relay tore down our SUBSCRIBE bidi) or publisherClosed.", + ) + } + if (firstSentButLost >= 0) { + fail( + "[$scope] publisher.send returned true for ALL sent frames, but listener missed " + + "${N_FRAMES - frames.size} of them starting at $firstSentButLost. " + + "Loss is downstream of moq-lite — uni-stream write threw (swallowed), " + + "QUIC flow control wedged, or relay dropped the stream.", + ) + } + } finally { + runCatching { subscription.unsubscribe() } + runCatching { publisher.close() } + runCatching { speakerWt.close(0, "test done") } + runCatching { listener.close() } + supervisor.cancelAndJoin() + } + Unit + } + + companion object { + private const val N_FRAMES = 100 + private const val FRAME_CADENCE_MS = 20L + private const val SUBSCRIBE_SETTLE_MS = 500L + private const val RECEIVE_TIMEOUT_MS = 30_000L + + private var harnessOrNull: NostrNestsHarness? = null + + @BeforeClass + @JvmStatic + fun setUpHarness() { + if (NostrNestsHarness.isEnabled()) { + harnessOrNull = NostrNestsHarness.shared() + } + } + + @AfterClass + @JvmStatic + fun tearDownHarness() { + harnessOrNull = null + } + } +} diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt index 677c10026..e919054ec 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt @@ -569,6 +569,188 @@ class NostrnestsProdAudioTransmissionTest { Unit } + /** + * Instrumented variant of [sustained_real_time_cadence_two_users] + * that BYPASSES [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster] + * and calls [com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle.send] + * directly so we can capture per-frame send results. + * + * Motivation: the production broadcaster wraps every send in + * `runCatching { publisher.send(opus); publisher.endGroup() }.onFailure {…}` + * and IGNORES the boolean. `send` returns false when + * `publisherClosed` or `inboundSubs.isEmpty()`, AND + * `runCatching { uni.write(framed) }` swallows write failures while + * still returning true. Frame loss is structurally invisible to the + * broadcaster. + * + * This test reproduces the speaker side WITHOUT the broadcaster: + * - same auth (OkHttpNestsClient.mintToken) + * - same transport (QuicWebTransportFactory) + * - same MoqLiteSession.client + session.publish() the + * broadcaster uses internally + * - records [MoqLitePublisherHandle.send]'s Boolean per frame and + * prints it alongside the listener's gap pattern + * + * Diagnostic table interpretation (sent vs. received per group): + * - send=false → speaker dropped the frame at the moq-lite layer + * (no inbound subscribers OR publisher closed) + * - send=true, received=false → frame was queued by moq-lite, then + * lost downstream (uni-stream write swallowed an exception, OR + * QUIC flow control / relay dropped the stream) + * - send=true, received=true → frame arrived (control) + */ + @Test + fun sustained_per_frame_send_outcomes_two_users() = + runBlocking { + assumeProd() + val scope = "send-trace" + + val hostSigner = NostrSignerInternal(KeyPair()) + val audienceSigner = NostrSignerInternal(KeyPair()) + val room = freshRoom(hostPubkey = hostSigner.pubKey) + val httpClient = OkHttpNestsClient { OkHttpClient() } + val transport = QuicWebTransportFactory() + + val supervisor = SupervisorJob() + val pumpScope = CoroutineScope(supervisor + Dispatchers.IO) + + InteropDebug.checkpoint( + scope, + "host=${hostSigner.pubKey.take(8)}… audience=${audienceSigner.pubKey.take(8)}… " + + "ns=${room.moqNamespace()} frames=$REAL_TIME_FRAMES cadence=${REAL_TIME_FRAME_MS}ms", + ) + + // ---- speaker side: build session manually so we can call + // session.publish() directly and capture per-frame send results. + val publishToken = + InteropDebug.stepSuspending(scope, "host: mintToken(publish=true)") { + httpClient.mintToken(room = room, publish = true, signer = hostSigner) + } + val (authority, path) = + com.vitorpamplona.nestsclient.buildRelayConnectTarget( + endpoint = room.endpoint, + namespace = room.moqNamespace(), + token = publishToken, + ) + val speakerWt = + InteropDebug.stepSuspending(scope, "host: WebTransport.connect") { + transport.connect(authority = authority, path = path, bearerToken = null) + } + val speakerSession = + com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession + .client(speakerWt, pumpScope) + val publisher = + InteropDebug.stepSuspending(scope, "host: session.publish(broadcastSuffix=hostPub)") { + speakerSession.publish(broadcastSuffix = hostSigner.pubKey) + } + + // ---- listener side: production code path, unchanged. + val listener = + InteropDebug.stepSuspending(scope, "audience: connectNestsListener") { + connectNestsListener( + httpClient = httpClient, + transport = transport, + scope = pumpScope, + room = room, + signer = audienceSigner, + ) + } + InteropDebug.assertListenerReached(scope, "Connected", listener.state.value) + + val subscription = + InteropDebug.stepSuspending(scope, "audience: subscribeSpeaker(host)") { + listener.subscribeSpeaker(hostSigner.pubKey) + } + + data class Arrival( + val wallMs: Long, + val groupId: Long, + ) + val received = java.util.concurrent.CopyOnWriteArrayList() + val sendOutcomes = BooleanArray(REAL_TIME_FRAMES) + val endGroupErrors = arrayOfNulls(REAL_TIME_FRAMES) + val collectStart = System.currentTimeMillis() + val collected = + async(pumpScope.coroutineContext) { + withTimeoutOrNull(REAL_TIME_RECEIVE_TIMEOUT_MS) { + subscription.objects + .onEach { obj -> + received += Arrival(System.currentTimeMillis() - collectStart, obj.groupId) + }.take(REAL_TIME_FRAMES) + .toList() + } + } + + try { + delay(SUBSCRIBE_SETTLE_MS) + + val payloadPrefix = ByteArray(79) { 0x4F.toByte() } + byteArrayOf(0x00) + val started = System.currentTimeMillis() + for (i in 0 until REAL_TIME_FRAMES) { + val payload = payloadPrefix + byteArrayOf(i.toByte()) + sendOutcomes[i] = publisher.send(payload) + runCatching { publisher.endGroup() } + .onFailure { endGroupErrors[i] = it::class.simpleName + ": " + it.message } + delay(REAL_TIME_FRAME_MS) + } + val pumpDurationMs = System.currentTimeMillis() - started + + collected.await() + val frames = received.toList() + val receivedGroups = frames.map { it.groupId }.toHashSet() + + val sendCount = sendOutcomes.count { it } + val firstFalseSend = sendOutcomes.indexOfFirst { !it } + val sendOnlyMissing = (0 until REAL_TIME_FRAMES).filter { sendOutcomes[it] && it.toLong() !in receivedGroups } + val firstUnreceivedAfterSend = sendOnlyMissing.firstOrNull() ?: -1 + + InteropDebug.checkpoint( + scope, + "sendTrue=$sendCount/$REAL_TIME_FRAMES received=${frames.size}/$REAL_TIME_FRAMES " + + "firstFalseSend=$firstFalseSend " + + "firstSentButLost=$firstUnreceivedAfterSend " + + "pumpDuration=${pumpDurationMs}ms", + ) + // Per-frame table: SEND TRUE / FALSE + RECV YES / NO + for (i in 0 until REAL_TIME_FRAMES) { + val sentOk = sendOutcomes[i] + val recv = i.toLong() in receivedGroups + val tag = + when { + sentOk && recv -> "ok" + sentOk && !recv -> "SENT-LOST" + !sentOk && !recv -> "send=false" + else -> "ghost?" + } + val err = endGroupErrors[i]?.let { ", endGroup err=$it" } ?: "" + InteropDebug.checkpoint(scope, " i=$i $tag$err") + } + + if (firstFalseSend >= 0) { + fail( + "[$scope] publisher.send returned false starting at frame $firstFalseSend — " + + "speaker's MoqLiteSession dropped the frame at the source. " + + "Likely cause: inboundSubs cleared (relay tore down our SUBSCRIBE bidi) or publisherClosed.", + ) + } + if (firstUnreceivedAfterSend >= 0) { + fail( + "[$scope] publisher.send returned true for ALL frames, but listener missed " + + "${REAL_TIME_FRAMES - frames.size} of them starting at $firstUnreceivedAfterSend. " + + "Loss is downstream of moq-lite — uni-stream write threw (swallowed by runCatching), " + + "QUIC flow control wedged, or the relay dropped the stream.", + ) + } + } finally { + runCatching { subscription.unsubscribe() } + runCatching { publisher.close() } + runCatching { speakerWt.close(0, "test done") } + runCatching { listener.close() } + supervisor.cancelAndJoin() + } + Unit + } + // ----- helpers ----- private fun freshRoom(hostPubkey: String): NestsRoomConfig =