From 2b86d71c1da6d4b31250c40de52539c6ab37e1a7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 21:22:53 +0000 Subject: [PATCH] test(audio-rooms): one listener's unsubscribe doesn't tear down others MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One speaker, two listeners A and B. Push 3 frames (both receive), A unsubscribes, push 3 more frames, B's stream completes with all 6. If A's unsubscribe accidentally tore down the speaker's broadcast or B's subscribe (e.g. an over-eager session-wide cleanup in a future refactor), B would time out waiting for the second batch — and the test names that exact failure mode in the message. --- .../NostrNestsUnsubscribeInteropTest.kt | 197 ++++++++++++++++++ 1 file changed, 197 insertions(+) create mode 100644 nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsUnsubscribeInteropTest.kt diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsUnsubscribeInteropTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsUnsubscribeInteropTest.kt new file mode 100644 index 000000000..a65bbe781 --- /dev/null +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsUnsubscribeInteropTest.kt @@ -0,0 +1,197 @@ +/* + * 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 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.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull +import org.junit.AfterClass +import org.junit.BeforeClass +import org.junit.Test +import kotlin.test.fail + +/** + * One speaker, two listeners. Listener A unsubscribes mid-stream; + * listener B keeps receiving frames. + * + * Pins the relay's "scoped to one subscription id" teardown — a + * common UI flow ("leave room while others stay") that should NOT + * tear down the speaker's broadcast or any other subscriber's flow. + * + * Skipped by default — set `-DnestsInterop=true` to enable. + */ +class NostrNestsUnsubscribeInteropTest { + @Test + fun unsubscribe_one_listener_keeps_other_subscriptions_alive() = + runBlocking { + NostrNestsHarness.assumeNestsInterop() + val harness = harnessOrNull ?: return@runBlocking + + val speakerSigner = NostrSignerInternal(KeyPair()) + val listenerSignerA = NostrSignerInternal(KeyPair()) + val listenerSignerB = NostrSignerInternal(KeyPair()) + val room = + NestsRoomConfig( + authBaseUrl = harness.authBaseUrl, + endpoint = harness.moqEndpoint, + hostPubkey = speakerSigner.pubKey, + roomId = "unsub-${System.nanoTime()}", + ) + val supervisor = SupervisorJob() + val pumpScope = CoroutineScope(supervisor + Dispatchers.IO) + val capture = InteropDriverCapture() + + val scope = "unsub" + InteropDebug.checkpoint(scope, "room=${room.moqNamespace()} speaker=${speakerSigner.pubKey.take(8)}") + + try { + val speaker = + InteropDebug.stepSuspending("$scope/speaker", "connectNestsSpeaker") { + connectNestsSpeaker( + httpClient = httpClient, + transport = transport, + scope = pumpScope, + room = room, + signer = speakerSigner, + speakerPubkeyHex = speakerSigner.pubKey, + captureFactory = { capture }, + encoderFactory = { InteropStubEncoder("FRAME-".encodeToByteArray()) }, + ) + } + val broadcast = + InteropDebug.stepSuspending("$scope/speaker", "startBroadcasting") { + speaker.startBroadcasting() + } + val listenerA = + InteropDebug.stepSuspending("$scope/listenerA", "connectNestsListener") { + connectNestsListener(httpClient, transport, pumpScope, room, listenerSignerA) + } + val listenerB = + InteropDebug.stepSuspending("$scope/listenerB", "connectNestsListener") { + connectNestsListener(httpClient, transport, pumpScope, room, listenerSignerB) + } + val subA = + InteropDebug.stepSuspending("$scope/listenerA", "subscribeSpeaker") { + listenerA.subscribeSpeaker(speakerSigner.pubKey) + } + val subB = + InteropDebug.stepSuspending("$scope/listenerB", "subscribeSpeaker") { + listenerB.subscribeSpeaker(speakerSigner.pubKey) + } + + // A wants only the first 3; B wants all 6. + val collectedA = + async(pumpScope.coroutineContext) { + withTimeoutOrNull(RECEIVE_TIMEOUT_MS) { + subA.objects.take(EARLY_FRAMES).toList() + } + } + val collectedB = + async(pumpScope.coroutineContext) { + withTimeoutOrNull(RECEIVE_TIMEOUT_MS) { + subB.objects.take(EARLY_FRAMES + LATE_FRAMES).toList() + } + } + delay(SUBSCRIBE_SETTLE_MS) + + InteropDebug.checkpoint(scope, "phase 1: push $EARLY_FRAMES frames (both A + B receive)") + for (i in 0 until EARLY_FRAMES) { + capture.push(i) + delay(FRAME_SPACING_MS) + } + + val resA = collectedA.await() + if (resA == null) { + fail("[$scope] A timed out waiting for $EARLY_FRAMES frames within ${RECEIVE_TIMEOUT_MS}ms") + } + InteropDebug.stepSuspending("$scope/listenerA", "unsubscribe") { subA.unsubscribe() } + delay(UNSUBSCRIBE_SETTLE_MS) + + InteropDebug.checkpoint(scope, "phase 2: push $LATE_FRAMES more frames (only B should receive)") + for (i in 0 until LATE_FRAMES) { + capture.push(EARLY_FRAMES + i) + delay(FRAME_SPACING_MS) + } + + val resB = collectedB.await() + if (resB == null) { + fail( + "[$scope] B timed out waiting for ${EARLY_FRAMES + LATE_FRAMES} frames " + + "within ${RECEIVE_TIMEOUT_MS}ms — A's unsubscribe should NOT have torn down B's stream", + ) + } + InteropDebug.checkpoint(scope, "A=${resA.size} frames, B=${resB.size} frames") + + runCatching { subB.unsubscribe() } + runCatching { listenerA.close() } + runCatching { listenerB.close() } + runCatching { broadcast.close() } + runCatching { speaker.close() } + } finally { + capture.stop() + supervisor.cancelAndJoin() + } + Unit + } + + companion object { + private const val EARLY_FRAMES = 3 + private const val LATE_FRAMES = 3 + private const val SUBSCRIBE_SETTLE_MS = 500L + private const val UNSUBSCRIBE_SETTLE_MS = 200L + private const val FRAME_SPACING_MS = 25L + private const val RECEIVE_TIMEOUT_MS = 10_000L + + private val httpClient = OkHttpNestsClient() + private val transport = + QuicWebTransportFactory(certificateValidator = PermissiveCertificateValidator()) + private var harnessOrNull: NostrNestsHarness? = null + + @BeforeClass + @JvmStatic + fun setUpHarness() { + if (NostrNestsHarness.isEnabled()) { + harnessOrNull = NostrNestsHarness.shared() + } + } + + @AfterClass + @JvmStatic + fun tearDownHarness() { + harnessOrNull = null + } + } +}