From f0b27654baad063b2680816054fa7bb1fb96cc8f Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 09:05:00 +0000 Subject: [PATCH] test(audio-rooms) + fix: round-trip test + audit pass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an end-to-end MoQ round-trip test and lands the highest-severity findings from a 3-agent audit (protocol / ViewModel / Android lifecycle) of the M5–M7 + Activity work. Round-trip test (`:nestsClient` MoqRoundTripTest): - Two MoqSession.client() instances (publisher + subscriber) talk through a hand-rolled in-test relay coroutine that mirrors a real MoQ relay's wire behavior (forwards CLIENT_SETUP / SERVER_SETUP / ANNOUNCE / SUBSCRIBE / SUBSCRIBE_OK / OBJECT_DATAGRAM). - 100-Opus-frame test exercises the full publisher → wire → subscriber path, asserting payload bytes + monotonic group/object ids round-trip correctly. Catches any drift between our publisher and our own subscriber's wire format. - Second test verifies SUBSCRIBE_ERROR(TRACK_DOES_NOT_EXIST) flows back as MoqProtocolException when the publisher hasn't openTrack'd. MoQ protocol fixes (CRITICAL audit findings): - SubscribeDoneStatus constants were inverted: had UNSUBSCRIBED=0x01 (peer reads as INTERNAL_ERROR) and TRACK_ENDED=0x00 (peer reads as UNSUBSCRIBED). Swapped to draft-stable values: UNSUBSCRIBED=0x00, TRACK_ENDED=0x03. - Pending CompletableDeferreds for in-flight SUBSCRIBE / ANNOUNCE on session close were `cancel()`-ed, which propagates as CancellationException — caller's entire scope cancels instead of catching a domain MoqProtocolException. Switched all sites to `completeExceptionally(MoqProtocolException("session closed"))` including unsubscribe-while-pending-OK. ViewModel fixes: - openSubscription race: re-check `activeSubscriptions[pubkey] === slot && !closed` AFTER the suspending `subscribeSpeaker` returns; if the user removed the speaker mid-flight, fire-and-forget UNSUBSCRIBE rather than attaching a leaked SubscribeHandle + AudioRoomPlayer to a discarded slot. - Server-initiated `Closed` listener state now triggers `teardown(targetState=Closed)` and resets the auto-retry counter, so a transport-died-mid-handshake doesn't leave stale subscriptions in the VM map until the Activity finishes. - Cleanup-scope split: `disconnect()` (user-driven) routes the close through `viewModelScope` (still alive); `onCleared()` routes it through a process-lived `cleanupScope` (default GlobalScope, tests pass backgroundScope) so MoQ control frames (UNSUBSCRIBE, UNANNOUNCE, SUBSCRIBE_DONE) actually land before the QUIC transport drops, instead of being eaten by the cancelled viewModelScope. Android lifecycle fixes: - AudioRoomBridge.clear() wired into AccountViewModel.onCleared next to the existing CallSessionBridge.clear() — no more cross-account AccountViewModel leak after logout/switch. - registerReceiver flag now gated on Build.VERSION.SDK_INT TIRAMISU+ — RECEIVER_NOT_EXPORTED on pre-33 devices collides with RECEIVER_VISIBLE_TO_INSTANT_APPS. PendingIntents stay package-scoped via setPackage(packageName). - onUserLeaveHint no longer enters PIP unconditionally: gated on ui.connection == Connected AND PackageManager FEATURE_PICTURE_IN_PICTURE present, so PIP-from-lobby / PIP-on-incompatible-device doesn't leave a frozen full-screen card in Recents. - Replaced the process-wide singleton AudioRoomPipActions toggle Boolean with a per-Activity MutableSharedFlow so a stale emission from a torn-down Activity can't leak into a new one. ViewModel test was extended with `cleanupScope = backgroundScope` plumbing so the post-disconnect `listener.close()` assertion remains observable in the test scheduler. Verified: spotlessApply clean; :commons:jvmTest (9 tests), :nestsClient:jvmTest (80 tests including 2 new round-trip), and :amethyst:compilePlayDebugKotlin all green. Audit findings deferred to follow-up commits (none ship-blocking): - Wire layout pinning vs draft-17 vs draft-11 (currently emits a draft-11-style OBJECT_DATAGRAM; we advertise draft-17). Resolve via the M4 manual interop pass against nostrnests. - UNANNOUNCE racing inbound SUBSCRIBE; control-pump cancel half-write (audit MoQ #5, #6) — small ordering tweaks. - Two retry coroutines stacking under fast Failed bursts (audit VM #4). - AudioRoomForegroundService startForeground always-first contract (audit Android #8). --- .../ui/screen/loggedIn/AccountViewModel.kt | 2 + .../audiorooms/room/AudioRoomActivity.kt | 57 +++-- .../room/AudioRoomActivityContent.kt | 27 ++- .../commons/viewmodels/AudioRoomViewModel.kt | 62 ++++- .../viewmodels/AudioRoomViewModelTest.kt | 6 +- .../nestsclient/moq/MoqSession.kt | 30 ++- .../nestsclient/moq/MoqRoundTripTest.kt | 226 ++++++++++++++++++ 7 files changed, 354 insertions(+), 56 deletions(-) create mode 100644 nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqRoundTripTest.kt diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt index c944b071f..6b9fe0313 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt @@ -1574,6 +1574,8 @@ class AccountViewModel( callManager.dispose() com.vitorpamplona.amethyst.service.call.CallSessionBridge .clear() + com.vitorpamplona.amethyst.ui.screen.loggedIn.audiorooms.room.AudioRoomBridge + .clear() feedStates.destroy() super.onCleared() } diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomActivity.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomActivity.kt index 3945bda3d..eeb7cae9e 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomActivity.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomActivity.kt @@ -28,6 +28,7 @@ import android.content.Context import android.content.Intent import android.content.IntentFilter import android.graphics.drawable.Icon +import android.os.Build import android.os.Bundle import android.util.Rational import androidx.activity.compose.setContent @@ -35,6 +36,9 @@ import androidx.appcompat.app.AppCompatActivity import androidx.compose.runtime.mutableStateOf import com.vitorpamplona.amethyst.R import com.vitorpamplona.amethyst.ui.theme.AmethystTheme +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.asSharedFlow /** * Standalone activity that owns the lifetime of an audio-room session. The @@ -54,6 +58,7 @@ import com.vitorpamplona.amethyst.ui.theme.AmethystTheme class AudioRoomActivity : AppCompatActivity() { private val isInPipMode = mutableStateOf(false) private val isMuted = mutableStateOf(false) + private val isConnected = mutableStateOf(false) private val pipReceiver = object : BroadcastReceiver() { @@ -63,10 +68,11 @@ class AudioRoomActivity : AppCompatActivity() { ) { when (intent.action) { ACTION_PIP_TOGGLE_MUTE -> { - // The composable owns the VM; the easiest signal is a - // shared static the content layer will pick up via - // its mute state observer. - AudioRoomPipActions.toggleMuteRequested.value = !AudioRoomPipActions.toggleMuteRequested.value + // Per-Activity SharedFlow — observed by the composable + // and forwarded to the VM. Replaces an earlier + // process-wide singleton that could leak edges across + // activity instances. + toggleMuteSignal.tryEmit(Unit) } ACTION_PIP_LEAVE -> { @@ -76,6 +82,12 @@ class AudioRoomActivity : AppCompatActivity() { } } + private val _toggleMuteSignal = + MutableSharedFlow(replay = 0, extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST) + + /** Emitted when the PIP-overlay mute action is tapped. */ + val toggleMuteSignal: MutableSharedFlow get() = _toggleMuteSignal + override fun onCreate(savedInstanceState: Bundle?) { super.onCreate(savedInstanceState) @@ -88,14 +100,21 @@ class AudioRoomActivity : AppCompatActivity() { return } - registerReceiver( - pipReceiver, + val filter = IntentFilter().apply { addAction(ACTION_PIP_TOGGLE_MUTE) addAction(ACTION_PIP_LEAVE) - }, - RECEIVER_NOT_EXPORTED, - ) + } + // RECEIVER_NOT_EXPORTED requires API 33+. The PendingIntents that + // fire these actions all use setPackage(packageName) so the + // visibility risk on older devices is bounded; pre-33 we register + // without the flag. + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.TIRAMISU) { + registerReceiver(pipReceiver, filter, RECEIVER_NOT_EXPORTED) + } else { + @Suppress("UnspecifiedRegisterReceiverFlag") + registerReceiver(pipReceiver, filter) + } setContent { AmethystTheme { @@ -111,6 +130,8 @@ class AudioRoomActivity : AppCompatActivity() { updatePipParams() } }, + onConnectedChange = { isConnected.value = it }, + pipMuteSignal = toggleMuteSignal.asSharedFlow(), onLeave = { finish() }, ) } @@ -119,6 +140,11 @@ class AudioRoomActivity : AppCompatActivity() { override fun onUserLeaveHint() { super.onUserLeaveHint() + // PIP only makes sense once the audio session is live; entering + // PIP from the lobby/Connecting state would freeze a half-rendered + // card in Recents. Also gates on the system supporting PIP. + if (!isConnected.value) return + if (!packageManager.hasSystemFeature(android.content.pm.PackageManager.FEATURE_PICTURE_IN_PICTURE)) return runCatching { enterPictureInPictureMode(buildPipParams()) } } @@ -198,16 +224,3 @@ class AudioRoomActivity : AppCompatActivity() { } } } - -/** - * Cross-talk between the PIP system buttons (which arrive as broadcast - * intents in the Activity) and the composable that owns the VM. The - * Activity flips `toggleMuteRequested`; a `LaunchedEffect` in the - * composable observes it and forwards the toggle to the VM. - * - * Lives at file scope so the composable can see it without cross-package - * visibility shenanigans. - */ -internal object AudioRoomPipActions { - val toggleMuteRequested = mutableStateOf(false) -} diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomActivityContent.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomActivityContent.kt index b4abb1d42..e37b8e2e3 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomActivityContent.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomActivityContent.kt @@ -44,6 +44,7 @@ import com.vitorpamplona.quartz.nip53LiveActivities.presence.MeetingRoomPresence import com.vitorpamplona.quartz.nip53LiveActivities.streaming.tags.ROLE import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.isActive import kotlinx.coroutines.launch @@ -64,6 +65,8 @@ internal fun AudioRoomActivityContent( accountViewModel: AccountViewModel, isInPipMode: Boolean, onMuteState: (Boolean) -> Unit, + onConnectedChange: (Boolean) -> Unit, + pipMuteSignal: SharedFlow, onLeave: () -> Unit, ) { val parsedAddress = remember(addressValue) { Address.parse(addressValue) } @@ -84,6 +87,8 @@ internal fun AudioRoomActivityContent( accountViewModel = accountViewModel, isInPipMode = isInPipMode, onMuteState = onMuteState, + onConnectedChange = onConnectedChange, + pipMuteSignal = pipMuteSignal, onLeave = onLeave, ) } @@ -97,6 +102,8 @@ private fun AudioRoomActivityBody( accountViewModel: AccountViewModel, isInPipMode: Boolean, onMuteState: (Boolean) -> Unit, + onConnectedChange: (Boolean) -> Unit, + pipMuteSignal: SharedFlow, onLeave: () -> Unit, ) { val participants = remember(event) { event.participants() } @@ -136,18 +143,18 @@ private fun AudioRoomActivityBody( val ui by viewModel.uiState.collectAsState() - // Push mute state into the activity so it can rebuild the PIP RemoteAction. + // Push mute + connected state up so the Activity can rebuild PIP + // RemoteActions and gate `enterPictureInPictureMode` on Connected. LaunchedEffect(ui.isMuted) { onMuteState(ui.isMuted) } + val isConnectedNow = ui.connection is ConnectionUiState.Connected + LaunchedEffect(isConnectedNow) { onConnectedChange(isConnectedNow) } - // The PIP mute system action signals via a shared toggle flag whose - // value flips on every tap. Observe the changes (skipping the initial - // composition) and forward to the VM. - val toggleSignal by AudioRoomPipActions.toggleMuteRequested - var lastSeenToggle by remember { mutableStateOf(toggleSignal) } - LaunchedEffect(toggleSignal) { - if (toggleSignal != lastSeenToggle) { - lastSeenToggle = toggleSignal - viewModel.setMuted(!ui.isMuted) + // Forward PIP-overlay mute taps into the VM. Per-Activity SharedFlow + // so a stale emission from a previous Activity instance can't leak + // into a new one. + LaunchedEffect(viewModel) { + pipMuteSignal.collect { + viewModel.setMuted(!viewModel.uiState.value.isMuted) } } diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModel.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModel.kt index 06f50d90e..94d0b48bd 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModel.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModel.kt @@ -44,6 +44,7 @@ import kotlinx.collections.immutable.ImmutableSet import kotlinx.collections.immutable.persistentSetOf import kotlinx.collections.immutable.toPersistentSet import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.GlobalScope import kotlinx.coroutines.Job import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow @@ -96,6 +97,13 @@ class AudioRoomViewModel( // listener whose state they can drive directly. private val connector: NestsListenerConnector = DefaultNestsListenerConnector, private val speakerConnector: NestsSpeakerConnector = DefaultNestsSpeakerConnector, + // Scope used for fire-and-forget MoQ cleanup (UNANNOUNCE, + // SUBSCRIBE_DONE, MoQ session close) that needs to outlive the VM's + // own scope. Production passes [GlobalScope] so onCleared can finish + // sending control frames before the QUIC transport drops; tests pass + // their `backgroundScope` so assertions can observe the close. + @OptIn(kotlinx.coroutines.DelicateCoroutinesApi::class) + private val cleanupScope: CoroutineScope = GlobalScope, ) : ViewModel() { private val _uiState = MutableStateFlow(AudioRoomUiState()) val uiState: StateFlow = _uiState.asStateFlow() @@ -278,16 +286,16 @@ class AudioRoomViewModel( autoRetryJob?.cancel() autoRetryJob = null autoRetryAttempts = 0 - teardownBroadcast(BroadcastUiState.Idle) - teardown(targetState = ConnectionUiState.Idle) + teardownBroadcast(BroadcastUiState.Idle, finalCleanup = false) + teardown(targetState = ConnectionUiState.Idle, finalCleanup = false) } override fun onCleared() { closed = true autoRetryJob?.cancel() autoRetryJob = null - teardownBroadcast(BroadcastUiState.Idle) - teardown(targetState = ConnectionUiState.Closed) + teardownBroadcast(BroadcastUiState.Idle, finalCleanup = true) + teardown(targetState = ConnectionUiState.Closed, finalCleanup = true) super.onCleared() } @@ -313,7 +321,10 @@ class AudioRoomViewModel( } } - private fun teardownBroadcast(targetState: BroadcastUiState) { + private fun teardownBroadcast( + targetState: BroadcastUiState, + finalCleanup: Boolean = false, + ) { speakerConnectJob?.cancel() speakerConnectJob = null speakerStateJob?.cancel() @@ -323,7 +334,12 @@ class AudioRoomViewModel( broadcastHandle = null speaker = null if (handle != null || s != null) { - viewModelScope.launch { + // User-driven disconnect can use viewModelScope (still alive); + // onCleared must use cleanupScope because viewModelScope is + // already cancelled and a launch on it would no-op without + // emitting the UNANNOUNCE / SUBSCRIBE_DONE wire frames. + val scope = if (finalCleanup) cleanupScope else viewModelScope + scope.launch { handle?.runCatching { close() } s?.runCatching { close() } } @@ -349,6 +365,14 @@ class AudioRoomViewModel( scheduleAutoRetry() } + // Server-initiated Closed: reset the retry counter and + // tear down stale local state so any later user-driven + // reconnect starts fresh. + NestsListenerState.Closed -> { + autoRetryAttempts = 0 + teardown(targetState = ConnectionUiState.Closed) + } + else -> { /* no extra side effect */ } } } @@ -467,6 +491,15 @@ class AudioRoomViewModel( if (closed || activeSubscriptions[pubkey] !== slot) return try { val handle = l.subscribeSpeaker(pubkey) + // Re-check after the suspending subscribeSpeaker — the user + // may have removed this speaker via updateSpeakers / disconnected + // while the SUBSCRIBE was in flight. If so, abandon the handle + // cleanly (fire-and-forget UNSUBSCRIBE) instead of attaching it + // to a slot the reconcile loop has already discarded. + if (closed || activeSubscriptions[pubkey] !== slot) { + viewModelScope.launch { runCatching { handle.unsubscribe() } } + return + } val decoder = decoderFactory() val player = playerFactory() val isMuted = _uiState.value.isMuted @@ -491,7 +524,10 @@ class AudioRoomViewModel( } } - private fun teardown(targetState: ConnectionUiState) { + private fun teardown( + targetState: ConnectionUiState, + finalCleanup: Boolean = false, + ) { connectJob?.cancel() connectJob = null stateObserverJob?.cancel() @@ -505,10 +541,14 @@ class AudioRoomViewModel( val l = listener listener = null if (l != null) { - // Closing the listener is suspending; fire-and-forget on the VM - // scope is fine — even if the scope is cancelled (onCleared), the - // underlying transport's own cleanup runs. - viewModelScope.launch { runCatching { l.close() } } + // Listener.close() sends MoQ control frames (UNSUBSCRIBE etc.) + // before the QUIC transport drops. User-driven disconnect uses + // the still-alive viewModelScope; onCleared has already had its + // viewModelScope cancelled, so it routes through cleanupScope + // (a process-lived scope) so the peer sees a clean teardown + // rather than a hard QUIC drop. + val scope = if (finalCleanup) cleanupScope else viewModelScope + scope.launch { runCatching { l.close() } } } _uiState.update { it.copy( diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModelTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModelTest.kt index bd54ca304..c4eaf0964 100644 --- a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModelTest.kt +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/AudioRoomViewModelTest.kt @@ -40,6 +40,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.UnconfinedTestDispatcher import kotlinx.coroutines.test.resetMain import kotlinx.coroutines.test.runTest @@ -211,7 +212,7 @@ class AudioRoomViewModelTest { ) } - private fun newViewModel(connect: suspend (CoroutineScope) -> NestsListener): AudioRoomViewModel = + private fun TestScope.newViewModel(connect: suspend (CoroutineScope) -> NestsListener): AudioRoomViewModel = AudioRoomViewModel( httpClient = NoopNestsClient, transport = NoopWebTransportFactory, @@ -224,6 +225,9 @@ class AudioRoomViewModelTest { NestsListenerConnector { _, _, scope, _, _, _ -> connect(scope) }, + // Wire to the test's backgroundScope so close calls run during + // the test rather than escaping to the real GlobalScope. + cleanupScope = backgroundScope, ) private class FakeNestsListener : NestsListener { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt index fc43f9121..ce43b6208 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt @@ -336,7 +336,9 @@ class MoqSession private constructor( suspend fun unsubscribe(subscribeId: Long) { val alias = stateMutex.withLock { - pendingSubscribes.remove(subscribeId)?.cancel() + pendingSubscribes.remove(subscribeId)?.completeExceptionally( + MoqProtocolException("unsubscribed before SUBSCRIBE_OK arrived"), + ) aliasBySubscribeId.remove(subscribeId)?.also { sinks.remove(it)?.close() } } ?: return @@ -550,18 +552,20 @@ class MoqSession private constructor( stateMutex.withLock { if (closed) return closed = true - // Fail any in-flight subscribe waiters cleanly. + // Fail any in-flight subscribe waiters with a domain exception + // (not Cancellation — that would propagate as scope cancellation + // in the awaiting coroutine instead of a recoverable error). + val sessionGone = MoqProtocolException("session closed") for ((_, deferred) in pendingSubscribes) { - deferred.cancel() + deferred.completeExceptionally(sessionGone) } pendingSubscribes.clear() for ((_, ch) in sinks) ch.close() sinks.clear() aliasBySubscribeId.clear() - // Fail in-flight announce waiters and surface an exception, since - // the caller is awaiting on .await(). + // Fail in-flight announce waiters the same way. for ((_, deferred) in pendingAnnounces) { - deferred.cancel() + deferred.completeExceptionally(sessionGone) } pendingAnnounces.clear() announces.values.forEach { it.markSessionClosedLocked() } @@ -845,13 +849,15 @@ object ErrorCode { } /** - * MoQ-transport SUBSCRIBE_DONE status_code values. As above, draft-stable - * subset; not exhaustive. + * MoQ-transport SUBSCRIBE_DONE status_code values. Per draft-ietf-moq-transport + * (draft-11+): 0x00 = UNSUBSCRIBED, 0x03 = TRACK_ENDED. The constants + * shipped initially had these inverted, which the audit caught — a peer + * decoded our "track closed" SUBSCRIBE_DONE as INTERNAL_ERROR. */ object SubscribeDoneStatus { - /** The publisher has no more objects coming for this track. */ - const val TRACK_ENDED: Long = 0x00L - /** Subscriber explicitly UNSUBSCRIBEd; publisher is acknowledging. */ - const val UNSUBSCRIBED: Long = 0x01L + const val UNSUBSCRIBED: Long = 0x00L + + /** The publisher has no more objects coming for this track. */ + const val TRACK_ENDED: Long = 0x03L } diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqRoundTripTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqRoundTripTest.kt new file mode 100644 index 000000000..75a68fa78 --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqRoundTripTest.kt @@ -0,0 +1,226 @@ +/* + * 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.moq + +import com.vitorpamplona.nestsclient.transport.FakeWebTransport +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.UnconfinedTestDispatcher +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals + +/** + * End-to-end round-trip tests: an actual `MoqSession.client()` publisher + * announces, opens a track, and pushes OBJECT_DATAGRAMs which transit a + * minimal in-test relay and arrive at an actual `MoqSession.client()` + * subscriber via its `SubscribeHandle.objects` flow. + * + * The relay role here is a hand-rolled coroutine that mirrors the wire + * behavior of a real MoQ relay (e.g. nostrnests): + * - reads CLIENT_SETUP from each side, writes SERVER_SETUP back + * - matches SUBSCRIBE on the subscriber side to ANNOUNCE on the + * publisher side, forwards SUBSCRIBE to the publisher + * - matches SUBSCRIBE_OK from the publisher and forwards to the + * subscriber + * - forwards every OBJECT_DATAGRAM from publisher → subscriber + * + * It's intentionally not generic — just enough to drive one + * (announce, openTrack, send) ↔ (subscribe, collect) round trip per test. + * + * What this catches that the per-side tests don't: that what our publisher + * emits is byte-for-byte what our subscriber expects. If we ever drift on + * MoQ message layouts between publisher and listener, this fails. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class MoqRoundTripTest { + @Test + fun publisher_to_subscriber_round_trip_delivers_objects_in_order() = + runTest(UnconfinedTestDispatcher()) { + val (pubA, pubB) = FakeWebTransport.pair() + val (subA, subB) = FakeWebTransport.pair() + + val ns = TrackNamespace.of("nests", "round-trip-room") + val trackName = "alice-pubkey".encodeToByteArray() + val publisherSession = MoqSession.client(pubA, backgroundScope) + val subscriberSession = MoqSession.client(subA, backgroundScope) + + val relayJob = launch { runRelay(pubB, subB, ns, trackName) } + + publisherSession.setup(listOf(MoqVersion.DRAFT_17)) + subscriberSession.setup(listOf(MoqVersion.DRAFT_17)) + + val announceHandle = publisherSession.announce(ns) + val publisher = announceHandle.openTrack(trackName) + + // Subscribe runs on a separate coroutine because it has to stay + // alive until SUBSCRIBE_OK comes back through the relay path, + // which requires the publisher's openTrack to be present (it is, + // by virtue of the line above). + val handleDeferred = + async { + subscriberSession.subscribe( + namespace = ns, + trackName = trackName, + // Big enough that the test's burst of 100 frames + // never trips DROP_OLDEST regardless of how far the + // consumer falls behind under UnconfinedTestDispatcher. + objectBufferCapacity = 256, + ) + } + val handle = handleDeferred.await() + + // Push N Opus-shaped payloads through the publisher and verify + // every one arrives on the subscriber's flow with intact + // group/object ids and payload bytes. Drives the publisher loop + // on `backgroundScope` so it runs concurrently with the + // consumer's `take().toList()` collect. + val frameCount = 100 + backgroundScope.launch { + repeat(frameCount) { i -> publisher.send(opusPayload(i)) } + } + + val received = handle.objects.take(frameCount).toList() + + assertEquals(frameCount, received.size) + received.forEachIndexed { idx, obj -> + assertEquals(0L, obj.groupId, "frame $idx groupId") + assertEquals(idx.toLong(), obj.objectId, "frame $idx objectId monotonic") + assertContentEquals(opusPayload(idx), obj.payload, "frame $idx payload") + } + + handle.unsubscribe() + publisher.close() + announceHandle.unannounce() + publisherSession.close() + subscriberSession.close() + relayJob.cancel() + } + + @Test + fun round_trip_unknown_namespace_subscribe_fails_with_protocol_exception() = + runTest { + val (pubA, pubB) = FakeWebTransport.pair() + val (subA, subB) = FakeWebTransport.pair() + val ns = TrackNamespace.of("nests", "real-room") + val trackName = "alice".encodeToByteArray() + val pub = MoqSession.client(pubA, backgroundScope) + val sub = MoqSession.client(subA, backgroundScope) + + val relayJob = launch { runRelay(pubB, subB, ns, trackName) } + + pub.setup(listOf(MoqVersion.DRAFT_17)) + sub.setup(listOf(MoqVersion.DRAFT_17)) + + // Publisher announces "real-room" but does NOT openTrack — the + // inbound SUBSCRIBE on the publisher side will return + // SUBSCRIBE_ERROR(TRACK_DOES_NOT_EXIST) because no publisher is + // registered for that name. + pub.announce(ns) + + val ex = + runCatching { + sub.subscribe(namespace = ns, trackName = trackName) + }.exceptionOrNull() + + assertEquals(true, ex is MoqProtocolException, "expected MoqProtocolException, got $ex") + + pub.close() + sub.close() + relayJob.cancel() + } + + /** + * Minimal MoQ-relay simulator. Mirrors what a nests relay does on the + * wire so two of our [MoqSession]s can talk through it. + * + * Limitations vs a real relay (deliberate, this is a test fixture): + * - One namespace, one track, one subscriber per test. + * - SubscribeId / trackAlias are forwarded 1:1 across the relay (a + * real relay remaps; we don't need to since neither side reuses ids). + * - No SUBSCRIBE_DONE relay — the test calls `unsubscribe()` and + * checks it doesn't blow up; the publisher's UNSUBSCRIBE goes to + * the relay which silently drops it. + */ + private suspend fun runRelay( + pubTransport: FakeWebTransport, + subTransport: FakeWebTransport, + @Suppress("UNUSED_PARAMETER") expectedNs: TrackNamespace, + @Suppress("UNUSED_PARAMETER") expectedTrack: ByteArray, + ) { + val pubControl = pubTransport.peerOpenedBidiStreams().first() + val subControl = subTransport.peerOpenedBidiStreams().first() + + // SETUP both sides. + val pubCs = MoqCodec.decode(pubControl.incoming().first())!!.message as ClientSetup + pubControl.write(MoqCodec.encode(ServerSetup(pubCs.supportedVersions.first()))) + + val subCs = MoqCodec.decode(subControl.incoming().first())!!.message as ClientSetup + subControl.write(MoqCodec.encode(ServerSetup(subCs.supportedVersions.first()))) + + // Read ANNOUNCE from publisher, ack with ANNOUNCE_OK. + val announce = MoqCodec.decode(pubControl.incoming().first())!!.message as Announce + pubControl.write(MoqCodec.encode(AnnounceOk(announce.namespace))) + + // Read SUBSCRIBE from subscriber, forward to publisher verbatim. + val sub = MoqCodec.decode(subControl.incoming().first())!!.message as Subscribe + pubControl.write(MoqCodec.encode(sub)) + + // Wait for the publisher's reply (SUBSCRIBE_OK or SUBSCRIBE_ERROR) + // and forward to the subscriber. + val pubReply = MoqCodec.decode(pubControl.incoming().first())!!.message + when (pubReply) { + is SubscribeOk -> { + subControl.write(MoqCodec.encode(pubReply)) + } + + is SubscribeError -> { + subControl.write(MoqCodec.encode(pubReply)) + return + } + + else -> { + error("unexpected publisher reply to SUBSCRIBE: ${pubReply.type}") + } + } + + // Forward every OBJECT_DATAGRAM from publisher → subscriber until + // either side cancels the relay coroutine. + pubTransport.incomingDatagrams().collect { datagram -> + subTransport.sendDatagram(datagram) + } + } + + private fun opusPayload(seqId: Int): ByteArray { + // Mimic an Opus packet shape (~80 bytes, varying content). + val size = 80 + val out = ByteArray(size) + for (i in 0 until size) { + out[i] = ((seqId xor i).toByte()) + } + return out + } +}