Merge pull request #2677 from vitorpamplona/claude/audio-transmission-tests-zKzlB
Add prod nostrnests.com audio transmission diagnostic tests
This commit is contained in:
Vitor Pamplona
@@ -98,4 +98,10 @@ tasks.withType<Test>().configureEach {
|
|||||||
System.getProperty("nestsInteropMoqRev")?.let { systemProperty("nestsInteropMoqRev", it) }
|
System.getProperty("nestsInteropMoqRev")?.let { systemProperty("nestsInteropMoqRev", it) }
|
||||||
System.getProperty("nestsInteropExternal")?.let { systemProperty("nestsInteropExternal", it) }
|
System.getProperty("nestsInteropExternal")?.let { systemProperty("nestsInteropExternal", it) }
|
||||||
System.getProperty("nestsInteropDebug")?.let { systemProperty("nestsInteropDebug", 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) }
|
||||||
}
|
}
|
||||||
|
|||||||
+529
@@ -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
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user