From 9f1b32f40db59813e56fe54778cbcbd8ee5b2ee6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 1 May 2026 01:08:14 +0000 Subject: [PATCH] Add prod nostrnests.com audio transmission diagnostic tests Four progressively-narrower JVM tests that drive the production connectNestsSpeaker / connectNestsListener / OkHttpNestsClient / QuicWebTransportFactory / NestMoqLiteBroadcaster code paths against the real moq.nostrnests.com + moq-auth.nostrnests.com infrastructure (no Docker harness, no Nostr events) so we can isolate why audio is not flowing between two real users: 1. auth_minting_works_for_publish_and_listen_paths 2. same_user_speaker_and_listener_round_trip 3. two_users_speaker_publishes_listener_subscribes 4. sustained_real_time_cadence_two_users (~2s @ 20ms cadence) Skipped by default; opt in with -DnestsProd=true and (optionally) override URLs via -DnestsProdEndpoint=... / -DnestsProdAuth=... --- nestsClient/build.gradle.kts | 6 + .../NostrnestsProdAudioTransmissionTest.kt | 529 ++++++++++++++++++ 2 files changed, 535 insertions(+) create mode 100644 nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt diff --git a/nestsClient/build.gradle.kts b/nestsClient/build.gradle.kts index a83e115f7..28ea6e980 100644 --- a/nestsClient/build.gradle.kts +++ b/nestsClient/build.gradle.kts @@ -98,4 +98,10 @@ tasks.withType().configureEach { System.getProperty("nestsInteropMoqRev")?.let { systemProperty("nestsInteropMoqRev", it) } System.getProperty("nestsInteropExternal")?.let { systemProperty("nestsInteropExternal", it) } System.getProperty("nestsInteropDebug")?.let { systemProperty("nestsInteropDebug", it) } + // Opt-in for tests that hit the real nostrnests.com infrastructure + // (see NostrnestsProdAudioTransmissionTest). Forwarded the same way + // the harness flags are. + System.getProperty("nestsProd")?.let { systemProperty("nestsProd", it) } + System.getProperty("nestsProdEndpoint")?.let { systemProperty("nestsProdEndpoint", it) } + System.getProperty("nestsProdAuth")?.let { systemProperty("nestsProdAuth", it) } } diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt new file mode 100644 index 000000000..f64e3d64e --- /dev/null +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt @@ -0,0 +1,529 @@ +/* + * 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.connectNestsListener +import com.vitorpamplona.nestsclient.connectNestsSpeaker +import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal +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.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull +import okhttp3.OkHttpClient +import org.junit.Assume.assumeTrue +import org.junit.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertTrue +import kotlin.test.fail + +/** + * Production-relay diagnostic suite. Drives the EXACT same speaker / + * listener entry points the Amethyst app uses + * ([connectNestsSpeaker] + [connectNestsListener] + + * [OkHttpNestsClient] + [QuicWebTransportFactory]) — pointed at the + * real `nostrnests.com` infrastructure instead of the local Docker + * harness — so we can isolate WHERE audio transmission breaks between + * two real users. + * + * The four tests progressively narrow the failure surface: + * + * 1. [auth_minting_works_for_publish_and_listen_paths] — does + * `POST https://moq-auth.nostrnests.com/auth` accept our NIP-98 + * signature for both `publish=true` (speaker) and `publish=false` + * (audience)? Failures here are auth / DNS / TLS, NOT transport. + * 2. [same_user_speaker_and_listener_round_trip] — one keypair drives + * both sides. Sidesteps the question of how nostrnests gates + * `publish=true` cross-user, so a green here proves + * QUIC + WebTransport + moq-lite + announce/subscribe all work + * against the production cert + ALPN. + * 3. [two_users_speaker_publishes_listener_subscribes] — user A is + * host + speaker, user B (a different keypair) is the audience. + * This is the actual app scenario. If #2 passes and this fails, + * the bug is in cross-user authorisation / policy, not transport. + * 4. [sustained_real_time_cadence_two_users] — same pair as #3 but + * pushes 100 frames at the production 20 ms Opus cadence (~2 s of + * audio). Catches flow-control / late-subscriber / per-track + * buffer issues that only show under steady traffic. + * + * Reused from the production app (same wire, same code): + * - [OkHttpNestsClient] — NIP-98-signed `POST /auth` + * - [QuicWebTransportFactory] — pure-Kotlin QUIC v1 + HTTP/3 + WT, + * `JdkCertificateValidator` for the real Let's Encrypt-style cert + * - [connectNestsSpeaker] / [connectNestsListener] — full state + * machine including SUBSCRIBE_OK settle delay, per-frame group + * framing, etc. + * - [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster] — + * the encode-and-publish loop, identical to what the Android UI + * wires into `AudioRecordCapture` + `MediaCodecOpusEncoder`. + * + * Substitutes (only because the JVM has no `android.media.*`): + * - [InteropDriverCapture] for `AudioRecordCapture` — same + * [com.vitorpamplona.nestsclient.audio.AudioCapture] interface. + * - [InteropStubEncoder] for `MediaCodecOpusEncoder` — emits a + * deterministic byte-prefixed payload so we can verify integrity + * end-to-end. Real Opus packets are also byte-array opaque from + * the moq-lite layer's perspective; the relay never inspects the + * payload, so substituting the encoder doesn't change the wire + * behaviour we're testing. + * + * No Nostr events. Each test mints a synthetic [NestsRoomConfig] with + * a random `roomId` so tests don't collide; the host pubkey is the + * speaker's identity. nostrnests's auth sidecar SHOULD accept this + * (the JWT just authorises a namespace + publish flag) — if it + * doesn't, that's itself a finding. + * + * Skipped by default. Enable with `-DnestsProd=true`. Override URLs + * with `-DnestsProdEndpoint=...` and `-DnestsProdAuth=...` if you + * point at a staging deployment. + */ +class NostrnestsProdAudioTransmissionTest { + @Test + fun auth_minting_works_for_publish_and_listen_paths() = + runBlocking { + assumeProd() + val scope = "auth-only" + val signer = NostrSignerInternal(KeyPair()) + val room = freshRoom(hostPubkey = signer.pubKey) + val httpClient = OkHttpNestsClient { OkHttpClient() } + + InteropDebug.checkpoint( + scope, + "auth=${room.authBaseUrl} endpoint=${room.endpoint} ns=${room.moqNamespace()}", + ) + + val publishToken = + InteropDebug.stepSuspending(scope, "mintToken(publish=true)") { + httpClient.mintToken(room = room, publish = true, signer = signer) + } + assertTrue(publishToken.isNotBlank(), "publish JWT must be non-empty") + assertTrue(publishToken.count { it == '.' } == 2, "JWT must have header.payload.signature") + + val listenToken = + InteropDebug.stepSuspending(scope, "mintToken(publish=false)") { + httpClient.mintToken(room = room, publish = false, signer = signer) + } + assertTrue(listenToken.isNotBlank(), "listen JWT must be non-empty") + assertTrue(listenToken.count { it == '.' } == 2, "JWT must have header.payload.signature") + + InteropDebug.checkpoint(scope, "publish JWT=${publishToken.take(24)}…") + InteropDebug.checkpoint(scope, "listen JWT=${listenToken.take(24)}…") + Unit + } + + @Test + fun same_user_speaker_and_listener_round_trip() = + runBlocking { + assumeProd() + val scope = "same-user" + + val signer = NostrSignerInternal(KeyPair()) + val room = freshRoom(hostPubkey = signer.pubKey) + val httpClient = OkHttpNestsClient { OkHttpClient() } + // Default JdkCertificateValidator — production cert is + // real, no permissive validator like the Docker harness. + val transport = QuicWebTransportFactory() + + val supervisor = SupervisorJob() + val pumpScope = CoroutineScope(supervisor + Dispatchers.IO) + + val capture = InteropDriverCapture() + val encoder = InteropStubEncoder(prefix = "PROD-A-".encodeToByteArray()) + + try { + val speaker = + InteropDebug.stepSuspending(scope, "connectNestsSpeaker") { + connectNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = pumpScope, + room = room, + signer = signer, + speakerPubkeyHex = signer.pubKey, + captureFactory = { capture }, + encoderFactory = { encoder }, + ) + } + InteropDebug.assertSpeakerReached(scope, "Connected", speaker.state.value) + + val broadcast = + InteropDebug.stepSuspending(scope, "speaker.startBroadcasting") { + speaker.startBroadcasting() + } + InteropDebug.assertSpeakerReached(scope, "Broadcasting", speaker.state.value) + + val listener = + InteropDebug.stepSuspending(scope, "connectNestsListener") { + connectNestsListener( + httpClient = httpClient, + transport = transport, + scope = pumpScope, + room = room, + signer = signer, + ) + } + InteropDebug.assertListenerReached(scope, "Connected", listener.state.value) + + val subscription = + InteropDebug.stepSuspending(scope, "listener.subscribeSpeaker(self)") { + listener.subscribeSpeaker(signer.pubKey) + } + + // Start collecting BEFORE pushing — moq-lite group frames + // dropped before .objects.collect() runs are gone. + val received = + async(pumpScope.coroutineContext) { + withTimeoutOrNull(RECEIVE_TIMEOUT_MS) { + subscription.objects.take(N_FRAMES).toList() + } + } + + // Settle delay so SUBSCRIBE_OK is in flight before the + // publisher's first send (otherwise TrackPublisher.send + // returns false and the object id rolls back). + delay(SUBSCRIBE_SETTLE_MS) + + InteropDebug.checkpoint(scope, "pushing $N_FRAMES frames") + for (i in 0 until N_FRAMES) { + capture.push(byteValue = i) + delay(FRAME_SPACING_MS) + } + + val frames = + InteropDebug.stepSuspending(scope, "await $N_FRAMES frames") { + received.await() + } + if (frames == null) { + fail( + "[$scope] Did not receive $N_FRAMES moq-lite frames within ${RECEIVE_TIMEOUT_MS}ms — " + + "speaker=${InteropDebug.describe(speaker.state.value)}, " + + "listener=${InteropDebug.describe(listener.state.value)}", + ) + } + assertEquals(N_FRAMES, frames.size, "expected $N_FRAMES objects") + frames.forEachIndexed { idx, obj -> + assertContentEquals( + "PROD-A-".encodeToByteArray() + byteArrayOf(idx.toByte()), + obj.payload, + "payload at index $idx round-tripped through nostrnests.com", + ) + } + + runCatching { subscription.unsubscribe() } + runCatching { broadcast.close() } + runCatching { listener.close() } + runCatching { speaker.close() } + } finally { + capture.stop() + supervisor.cancelAndJoin() + } + Unit + } + + @Test + fun two_users_speaker_publishes_listener_subscribes() = + runBlocking { + assumeProd() + val scope = "two-users" + + val hostSigner = NostrSignerInternal(KeyPair()) + val audienceSigner = NostrSignerInternal(KeyPair()) + // Host pubkey defines the room namespace and the broadcast + // suffix. Audience uses a DIFFERENT keypair to mint a + // listen-only JWT for the same namespace — this is the + // exact app flow when user B joins user A's room. + val room = freshRoom(hostPubkey = hostSigner.pubKey) + + val httpClient = OkHttpNestsClient { OkHttpClient() } + val transport = QuicWebTransportFactory() + + val supervisor = SupervisorJob() + val pumpScope = CoroutineScope(supervisor + Dispatchers.IO) + + val capture = InteropDriverCapture() + val encoder = InteropStubEncoder(prefix = "PROD-B-".encodeToByteArray()) + + InteropDebug.checkpoint( + scope, + "host=${hostSigner.pubKey.take(8)}… audience=${audienceSigner.pubKey.take(8)}… " + + "ns=${room.moqNamespace()}", + ) + + try { + val speaker = + InteropDebug.stepSuspending(scope, "host: connectNestsSpeaker") { + connectNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = pumpScope, + room = room, + signer = hostSigner, + speakerPubkeyHex = hostSigner.pubKey, + captureFactory = { capture }, + encoderFactory = { encoder }, + ) + } + InteropDebug.assertSpeakerReached(scope, "Connected", speaker.state.value) + + val broadcast = + InteropDebug.stepSuspending(scope, "host: startBroadcasting") { + speaker.startBroadcasting() + } + InteropDebug.assertSpeakerReached(scope, "Broadcasting", speaker.state.value) + + 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) + } + + val received = + async(pumpScope.coroutineContext) { + withTimeoutOrNull(RECEIVE_TIMEOUT_MS) { + subscription.objects.take(N_FRAMES).toList() + } + } + delay(SUBSCRIBE_SETTLE_MS) + + InteropDebug.checkpoint(scope, "host pushing $N_FRAMES frames") + for (i in 0 until N_FRAMES) { + capture.push(byteValue = i) + delay(FRAME_SPACING_MS) + } + + val frames = + InteropDebug.stepSuspending(scope, "audience: await $N_FRAMES frames") { + received.await() + } + if (frames == null) { + fail( + "[$scope] audience did not receive $N_FRAMES frames within " + + "${RECEIVE_TIMEOUT_MS}ms — speaker=" + + InteropDebug.describe(speaker.state.value) + + ", listener=" + InteropDebug.describe(listener.state.value) + + ". If same-user round-trip passes but this fails, the bug is " + + "cross-user authorisation (nostrnests auth sidecar policy) or " + + "broadcast-suffix routing, NOT transport.", + ) + } + assertEquals(N_FRAMES, frames.size, "audience must see exactly $N_FRAMES objects") + frames.forEachIndexed { idx, obj -> + assertContentEquals( + "PROD-B-".encodeToByteArray() + byteArrayOf(idx.toByte()), + obj.payload, + "audience payload at index $idx must match what the host encoded", + ) + } + + runCatching { subscription.unsubscribe() } + runCatching { broadcast.close() } + runCatching { listener.close() } + runCatching { speaker.close() } + } finally { + capture.stop() + supervisor.cancelAndJoin() + } + Unit + } + + @Test + fun sustained_real_time_cadence_two_users() = + runBlocking { + assumeProd() + val scope = "two-users-sustained" + + 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) + + val capture = InteropDriverCapture() + // Realistic Opus payload size: ~80 bytes for a 20 ms / 32 + // kbps frame. The relay never inspects the bytes, but the + // size matters for stream / datagram path choices. + val payloadPrefix = ByteArray(79) { 0x4F.toByte() } + byteArrayOf(0x00) + val encoder = InteropStubEncoder(prefix = payloadPrefix) + + 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 " + + "(~${REAL_TIME_FRAMES * REAL_TIME_FRAME_MS}ms of audio)", + ) + + try { + val speaker = + InteropDebug.stepSuspending(scope, "host: connectNestsSpeaker") { + connectNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = pumpScope, + room = room, + signer = hostSigner, + speakerPubkeyHex = hostSigner.pubKey, + captureFactory = { capture }, + encoderFactory = { encoder }, + ) + } + InteropDebug.assertSpeakerReached(scope, "Connected", speaker.state.value) + val broadcast = speaker.startBroadcasting() + InteropDebug.assertSpeakerReached(scope, "Broadcasting", speaker.state.value) + + val listener = + connectNestsListener( + httpClient = httpClient, + transport = transport, + scope = pumpScope, + room = room, + signer = audienceSigner, + ) + InteropDebug.assertListenerReached(scope, "Connected", listener.state.value) + + val subscription = listener.subscribeSpeaker(hostSigner.pubKey) + + val received = + async(pumpScope.coroutineContext) { + withTimeoutOrNull(REAL_TIME_RECEIVE_TIMEOUT_MS) { + subscription.objects.take(REAL_TIME_FRAMES).toList() + } + } + delay(SUBSCRIBE_SETTLE_MS) + + val started = System.currentTimeMillis() + for (i in 0 until REAL_TIME_FRAMES) { + capture.push(byteValue = i and 0xFF) + delay(REAL_TIME_FRAME_MS) + } + val pumpDurationMs = System.currentTimeMillis() - started + + val frames = received.await() + if (frames == null) { + fail( + "[$scope] only got partial audio within " + + "${REAL_TIME_RECEIVE_TIMEOUT_MS}ms — speaker=" + + InteropDebug.describe(speaker.state.value) + + ", listener=" + InteropDebug.describe(listener.state.value) + + ", pumpDuration=${pumpDurationMs}ms. If two_users round-trip " + + "passes but this drops frames, suspect flow control / " + + "datagram drops / per-subscription buffer.", + ) + } + // Expect every frame: the broadcaster opens one moq-lite + // group per frame, so there's no group-level dropping + // for late attach inside the 100-frame window. + assertEquals( + REAL_TIME_FRAMES, + frames.size, + "expected all $REAL_TIME_FRAMES sustained-cadence frames; received ${frames.size}", + ) + // groupId must be monotonic and dense (no gaps) — gap + // detection is the cheapest signal that the relay + // dropped a uni stream mid-flight. + frames.forEachIndexed { idx, obj -> + assertEquals( + idx.toLong(), + obj.groupId, + "groupId gap at index $idx — relay dropped a uni stream", + ) + } + + runCatching { subscription.unsubscribe() } + runCatching { broadcast.close() } + runCatching { listener.close() } + runCatching { speaker.close() } + } finally { + capture.stop() + supervisor.cancelAndJoin() + } + Unit + } + + // ----- helpers ----- + + private fun freshRoom(hostPubkey: String): NestsRoomConfig = + NestsRoomConfig( + authBaseUrl = System.getProperty(PROD_AUTH_PROPERTY) ?: DEFAULT_AUTH_URL, + endpoint = System.getProperty(PROD_ENDPOINT_PROPERTY) ?: DEFAULT_ENDPOINT_URL, + hostPubkey = hostPubkey, + roomId = "prod-diag-${System.currentTimeMillis()}-${(0..9999).random()}", + ) + + private fun assumeProd() { + val enabled = System.getProperty(PROD_PROPERTY) == "true" + assumeTrue( + "Skipping nostrnests.com production test — set -D$PROD_PROPERTY=true to enable. " + + "Override URLs with -D$PROD_ENDPOINT_PROPERTY=... and -D$PROD_AUTH_PROPERTY=...", + enabled, + ) + } + + companion object { + // Hardcoded in the Android app's NestsServersScreen — same + // values the production app ships with. + private const val DEFAULT_ENDPOINT_URL = "https://moq.nostrnests.com:4443" + private const val DEFAULT_AUTH_URL = "https://moq-auth.nostrnests.com" + + private const val PROD_PROPERTY = "nestsProd" + private const val PROD_ENDPOINT_PROPERTY = "nestsProdEndpoint" + private const val PROD_AUTH_PROPERTY = "nestsProdAuth" + + // Round-trip tunables — same shape as the Docker harness round + // trip, large enough to detect drops but small enough to keep + // the test under 30 s wall-clock against a real WAN relay. + private const val N_FRAMES = 8 + private const val SUBSCRIBE_SETTLE_MS = 500L + private const val FRAME_SPACING_MS = 25L + private const val RECEIVE_TIMEOUT_MS = 20_000L + + // Sustained cadence: ~2 s of audio at the production Opus + // frame size (20 ms). Same cadence the Android broadcaster + // hits in real life. + private const val REAL_TIME_FRAMES = 100 + private const val REAL_TIME_FRAME_MS = 20L + private const val REAL_TIME_RECEIVE_TIMEOUT_MS = 30_000L + } +}