From 0b1ec52f799b2af7e053c3d85e8ec54a5e5646ee Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 12:04:18 +0000 Subject: [PATCH] =?UTF-8?q?fix(audio-rooms):=20audit=20follow-up=20?= =?UTF-8?q?=E2=80=94=20MoQ=20HIGH/MED=20+=20VM=20concurrency=20+=20Android?= =?UTF-8?q?=20polish?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 of the audit-driven cleanup. Lands every HIGH and most MED findings from the protocol / ViewModel / Android lifecycle audits that weren't fixed in the previous commit. The wire-format draft pinning (audit MoQ #1, #8) is intentionally deferred until the M4 manual interop pass against `nostrnests.com` reveals which draft revision the relay is actually speaking. MoQ session fixes: - #5 UNANNOUNCE wire-write now happens BEFORE removing announces[ns], so an inbound SUBSCRIBE during the teardown window sees the namespace as withdrawn (sessionClosed=true → SUBSCRIBE_ERROR(TRACK_DOES_NOT_EXIST)) instead of "namespace never existed". - #4 dispatchControlMessage(AnnounceError) now distinguishes pre-OK and post-OK errors. Post-OK is a session-level kick: mark the handle closed, send UNANNOUNCE, then drop. Pre-OK still rolls back the optimistic announces[] insert as before. - #6 close() now joins the cancelled control + datagram pumps before calling controlStream.finish(), so an in-flight SUBSCRIBE_OK / SUBSCRIBE_DONE / SUBSCRIBE_ERROR write can complete its writeMutex.withLock { ... } critical section. Previously cancellation could truncate a frame mid-flight and we'd send FIN over a corrupted stream. - #9 pumps wrapped in try/catch that calls close(...) on unexpected exceptions, so a transport-died-mid-session no longer leaves the session thinking it's healthy with new subscribe/announce calls hanging on a dead peer. - #10 TrackPublisher.send rolls back nextObjectId when every datagram fan-out fails (transport down). The audio-rooms NIP wants strictly contiguous object ids per group; a gap from a fully-failed send would trip strict subscribers. - #13 DefaultNestsSpeaker.close drops `gate` before calling activeHandle.close() / session.close(). The teardown chain runs cancelAndJoin on the broadcaster + sends SUBSCRIBE_DONE per attached subscriber + joins MoQ pumps; holding the gate through all of that blocked any other concurrent API call on this speaker. Resource lifecycle (audit MoQ #11/#12): - AudioRoomBroadcaster.stop() now `cancelAndJoin`s the loop before releasing the encoder + closing the publisher. The loop's last encoder.encode/publisher.send no longer races encoder.release()/publisher.close() — both produced use-after-release on native MediaCodec on Android, the latter sent orphan OBJECT_DATAGRAMs to subscribers we'd just told SUBSCRIBE_DONE. - AudioRoomPlayer.stop() promoted to `suspend` + cancelAndJoin for the same reason: decoder.release() ran while the decode loop was still inside MediaCodec.decode(...), undefined behaviour. Updated VM call sites (closeSubscription, teardown) to route both player.stop() and handle.unsubscribe() through one launched coroutine via the new `detach(): Pair` shape. ViewModel fixes: - #4 auto-retry uses a single `retryPending: Boolean` flag instead of `Job.isActive`. Two scheduleAutoRetry calls could previously both pass `Job.isActive == false` (the launched body had just started) and stack a second retry on top of one already running. - #7 setMicMuted updates the UI INSIDE the launched coroutine, after the suspending broadcastHandle.setMuted() returns. Previously the indicator could claim "muted" while audio was still on the wire if the handle's setMuted suspended on a gate. - #8 connect() cancels the previous stateObserverJob before kicking off the new attempt, so a delayed Failed/Closed emission from the old listener can no longer clobber the fresh Connecting UI. - #9 disconnect() clears requestedSpeakers, so a fresh connect() to a different room (or the same room after a long pause) doesn't reuse a stale speaker snapshot. - #12 updateSpeakers filters out the user's own pubkey: subscribing to your own forwarded audio would echo through the local playback device whenever the broadcast track loops back from the relay. Android lifecycle / PIP / service: - #6 PIP aspect ratio flipped from 9:16 (portrait sliver) to 16:9 (landscape) so the row of avatars under the title actually fits. - #7 process-death recovery: when AudioRoomBridge is empty (previous process's AccountViewModel is gone), redirect to MainActivity before finish() so the user lands somewhere meaningful instead of a black-flash. - #8 AudioRoomForegroundService.onStartCommand always calls startForeground first, on every invocation including ACTION_STOP. startForegroundService's 5-second contract requires it; previously the STOP path skipped it. startForeground itself wrapped in runCatching so a foreground-not-allowed exception bails cleanly rather than leaking the wake-lock. - #10 wake-lock timeout reduced 12 h → 4 h. Stuck connections that fail to detect a network drop no longer hold the device awake for half a day. - #11 presence-event spam fix: split the publish loop into a heartbeat keyed only on (address, handRaised) and a separate debounced state-change publisher keyed on micMutedTag. Every mute toggle previously triggered a full sign + publish + relay round trip; now we coalesce within a 500 ms window. - #12 final "leaving" presence routed through GlobalScope.launch instead of rememberCoroutineScope (which is cancelled on dispose, so the leave event almost never reached the relay). - #14 RECORD_AUDIO denial recovery: when the user has tapped "Don't ask again", the launcher silently returns false. New "Open settings" button deep-links to the app's settings page so the user can re-grant the permission and try again. - #15 setPictureInPictureParams now updates outside PIP too, so the next entry shows the correct mute-state icon without an extra flip. - #16 onNewIntent override: a second Join tap for a different room finishes the current Activity and starts a fresh one with the new extras, instead of singleTask silently keeping the old room running. Verified: spotlessApply clean; :commons:jvmTest, :nestsClient:jvmTest (80 tests), :amethyst:compilePlayDebugKotlin all green. Audit findings still deferred (all documented inline / in this commit): - MoQ #1, #8: wire-format draft pinning (draft-17 vs draft-11). Needs the M4 manual interop pass to confirm what nests is actually speaking. - VM #6: confined map mutation under a single dispatcher. Current setup (Dispatchers.Main via setMain in tests, viewModelScope in prod) is functionally fine; full belt-and-suspenders confining is a separate concurrency review. - VM #10: test coverage gaps for retry + speaker reconcile cycle + setMicMuted no-handle case + server-initiated Closed leaves stale state. Each is a dedicated test. - Android #18: startListening / promoteToMicrophone race. Mitigation (always declaring microphone foreground type) requires unconditional RECORD_AUDIO grant which listener-only users won't have. --- .../audiorooms/AudioRoomForegroundService.kt | 35 +++-- .../audiorooms/room/AudioRoomActivity.kt | 36 ++++- .../room/AudioRoomActivityContent.kt | 24 ++- .../audiorooms/room/AudioRoomFullScreen.kt | 19 +++ amethyst/src/main/res/values/strings.xml | 1 + .../commons/viewmodels/AudioRoomViewModel.kt | 138 +++++++++++++----- .../vitorpamplona/nestsclient/NestsSpeaker.kt | 14 +- .../nestsclient/audio/AudioRoomBroadcaster.kt | 10 +- .../nestsclient/audio/AudioRoomPlayer.kt | 15 +- .../nestsclient/moq/MoqSession.kt | 107 ++++++++++++-- 10 files changed, 323 insertions(+), 76 deletions(-) diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/audiorooms/AudioRoomForegroundService.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/audiorooms/AudioRoomForegroundService.kt index ca3a73e4c..f11da1cc3 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/audiorooms/AudioRoomForegroundService.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/audiorooms/AudioRoomForegroundService.kt @@ -72,20 +72,30 @@ class AudioRoomForegroundService : Service() { flags: Int, startId: Int, ): Int { - when (intent?.action) { - ACTION_PROMOTE_TO_MIC -> { - startForegroundWithType(includeMic = true) + // Always call startForeground first — Android's contract is that + // every onStartCommand for a service started via + // startForegroundService MUST call startForeground within 5 s, on + // every invocation (audit Android #8). Even the STOP path needs a + // foreground state before stopForeground can demote it cleanly. + val mic = + when (intent?.action) { + ACTION_PROMOTE_TO_MIC -> true + else -> promoted } - - ACTION_STOP -> { - stopForeground(STOP_FOREGROUND_REMOVE) + runCatching { startForegroundWithType(includeMic = mic) } + .onFailure { + // foreground-not-allowed (mic-only restrictions, etc.) — + // bail cleanly rather than leak a wake-lock / partial state. stopSelf() + return START_NOT_STICKY } - else -> { - startForegroundWithType(includeMic = promoted) - } + if (intent?.action == ACTION_STOP) { + stopForeground(STOP_FOREGROUND_REMOVE) + stopSelf() + return START_NOT_STICKY } + if (wakeLock?.isHeld != true) { wakeLock?.acquire(WAKE_LOCK_TIMEOUT_MS) } @@ -173,7 +183,12 @@ class AudioRoomForegroundService : Service() { private const val NOTIFICATION_ID = 0xA0D10 private const val ACTION_PROMOTE_TO_MIC = "com.vitorpamplona.amethyst.audio_room.PROMOTE_MIC" private const val ACTION_STOP = "com.vitorpamplona.amethyst.audio_room.STOP" - private const val WAKE_LOCK_TIMEOUT_MS = 12L * 60 * 60 * 1000 // 12 hours hard cap + + // 4-hour hard cap — long enough for typical audio-room sessions + // (2-3 h podcasts / panels) plus headroom, short enough to release + // the device from a stuck connection that fails to detect a + // network drop. Refresh on each ACTION_PROMOTE_TO_MIC arrival. + private const val WAKE_LOCK_TIMEOUT_MS = 4L * 60 * 60 * 1000 /** Start the service in listener-only foreground mode. Idempotent. */ fun startListening(context: Context) { 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 eeb7cae9e..3fc0aeeca 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 @@ -96,6 +96,19 @@ class AudioRoomActivity : AppCompatActivity() { val serviceBase = intent.getStringExtra(EXTRA_SERVICE_BASE) val roomId = intent.getStringExtra(EXTRA_ROOM_ID) if (accountViewModel == null || addressValue == null || serviceBase == null || roomId == null) { + // After process death the bridge is empty (the previous + // process's AccountViewModel is gone). Bounce the user back to + // MainActivity so they land on the lobby instead of a black + // screen flashing closed (audit Android #7). + if (accountViewModel == null) { + runCatching { + startActivity( + Intent(this, com.vitorpamplona.amethyst.ui.MainActivity::class.java).apply { + addFlags(Intent.FLAG_ACTIVITY_CLEAR_TOP or Intent.FLAG_ACTIVITY_NEW_TASK) + }, + ) + } + } finish() return } @@ -148,20 +161,39 @@ class AudioRoomActivity : AppCompatActivity() { runCatching { enterPictureInPictureMode(buildPipParams()) } } + override fun onNewIntent(intent: Intent) { + super.onNewIntent(intent) + // singleTask brings an existing instance forward when the user + // taps Join from a different room. We finish + let the new + // intent's startActivity create a fresh activity rather than + // silently keeping the old room running (audit Android #16). + val newAddress = intent.getStringExtra(EXTRA_ADDRESS) + val currentAddress = getIntent()?.getStringExtra(EXTRA_ADDRESS) + if (newAddress != null && newAddress != currentAddress) { + finish() + startActivity(intent) + } + } + override fun onPictureInPictureModeChanged(isInPictureInPictureMode: Boolean) { super.onPictureInPictureModeChanged(isInPictureInPictureMode) isInPipMode.value = isInPictureInPictureMode } private fun updatePipParams() { - if (!isInPipMode.value) return + // Pre-stage params even outside PIP so the next entry shows the + // correct mute icon (audit Android #15). setPictureInPictureParams + // is legal in any state on API 26+. runCatching { setPictureInPictureParams(buildPipParams()) } } private fun buildPipParams(): PictureInPictureParams = PictureInPictureParams .Builder() - .setAspectRatio(Rational(9, 16)) + // Landscape ratio — the PIP layout is a horizontal row of + // avatars under a title (see AudioRoomPipScreen), which 9:16 + // squeezed into a sliver. 16:9 fits the row and the room name. + .setAspectRatio(Rational(16, 9)) .setActions(buildPipActions()) .build() 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 e37b8e2e3..a1f3604f1 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 @@ -167,16 +167,35 @@ private fun AudioRoomActivityBody( is BroadcastUiState.Broadcasting -> b.isMuted else -> null } - LaunchedEffect(event.address().toValue(), handRaised, micMutedTag) { + // Heartbeat loop — publishes once on enter / hand-raise change, then + // every 30 s. micMutedTag is intentionally NOT a key here: every mute + // toggle would otherwise trigger a presence publish + relay round trip + // (audit Android #11). The next heartbeat picks up the latest mic + // state up to 30 s later, which is well within the user's "did the + // peer see my mute" tolerance. + LaunchedEffect(event.address().toValue(), handRaised) { publishPresence(account, event, handRaised, micMutedTag) while (isActive) { delay(PRESENCE_REFRESH_MS) publishPresence(account, event, handRaised, micMutedTag) } } + // Debounced state-change publisher: after a mute toggle, wait + // PRESENCE_DEBOUNCE_MS for further changes before sending a fresh + // presence event. The LaunchedEffect's auto-cancel on key change + // serves as the debounce mechanism. + LaunchedEffect(micMutedTag) { + delay(PRESENCE_DEBOUNCE_MS) + publishPresence(account, event, handRaised, micMutedTag) + } DisposableEffect(event.address().toValue()) { onDispose { - scope.launch(Dispatchers.IO) { + // Final "leaving" presence runs on a non-cancellable scope so + // it survives the composable's scope being cancelled mid- + // network. Without this the leave event almost never reaches + // the relay (audit Android #12). + @OptIn(kotlinx.coroutines.DelicateCoroutinesApi::class) + kotlinx.coroutines.GlobalScope.launch(Dispatchers.IO) { runCatching { publishPresence(account, event, handRaised = false, micMuted = null) } } } @@ -219,6 +238,7 @@ private fun AudioRoomActivityBody( } private const val PRESENCE_REFRESH_MS = 30_000L +private const val PRESENCE_DEBOUNCE_MS = 500L private suspend fun publishPresence( account: com.vitorpamplona.amethyst.model.Account, diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomFullScreen.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomFullScreen.kt index 14247f3b0..c3fa9378a 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomFullScreen.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/audiorooms/room/AudioRoomFullScreen.kt @@ -258,6 +258,25 @@ private fun TalkRow( style = MaterialTheme.typography.bodySmall, color = MaterialTheme.colorScheme.error, ) + // After "Don't ask again" the permission launcher + // silently returns false. Give the user a path to + // toggle the permission in system settings (audit + // Android #14). + OutlinedButton(onClick = { + runCatching { + context.startActivity( + android.content + .Intent( + android.provider.Settings.ACTION_APPLICATION_DETAILS_SETTINGS, + android.net.Uri.fromParts("package", context.packageName, null), + ).apply { + addFlags(android.content.Intent.FLAG_ACTIVITY_NEW_TASK) + }, + ) + } + }) { + Text(stringRes(R.string.audio_room_open_settings)) + } } } diff --git a/amethyst/src/main/res/values/strings.xml b/amethyst/src/main/res/values/strings.xml index 54edeadfe..b432454c8 100644 --- a/amethyst/src/main/res/values/strings.xml +++ b/amethyst/src/main/res/values/strings.xml @@ -512,6 +512,7 @@ Live Broadcast failed: %1$s Microphone access is required to talk in this room. + Open settings Audio rooms Keeps audio playing while a room is open. Audio room connected 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 94d0b48bd..d30614e3b 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 @@ -123,16 +123,28 @@ class AudioRoomViewModel( private var autoRetryAttempts = 0 private var autoRetryJob: Job? = null + // True between scheduleAutoRetry() and the launched coroutine's + // `finally`. Single source of truth — Job.isActive flips to false + // the moment the launched body starts running, which created a + // window where two retries could both pass the gate. + private var retryPending = false + // Speaker / publisher path private var speaker: NestsSpeaker? = null private var broadcastHandle: BroadcastHandle? = null private var speakerStateJob: Job? = null private var speakerConnectJob: Job? = null - /** Push the latest known speaker set from the room event. */ + /** + * Push the latest known speaker set from the room event. The user's + * own pubkey (when broadcasting) is filtered out so we don't subscribe + * to our own forwarded audio — that would echo through the local + * playback device whenever our broadcast track loops back from the relay. + */ fun updateSpeakers(speakerPubkeys: Set) { if (closed) return - requestedSpeakers = speakerPubkeys + val selfHex = signer.pubKey + requestedSpeakers = if (selfHex.isEmpty()) speakerPubkeys else speakerPubkeys - selfHex if (_uiState.value.connection is ConnectionUiState.Connected) { reconcileSubscriptions() } @@ -151,6 +163,12 @@ class AudioRoomViewModel( autoRetryJob?.cancel() autoRetryJob = null autoRetryAttempts = 0 + retryPending = false + // Cancel any previous state observer too — otherwise its delayed + // emissions can clobber the fresh Connecting UI we set below + // (audit VM #8). + stateObserverJob?.cancel() + stateObserverJob = null _uiState.update { it.copy(connection = ConnectionUiState.Connecting(ConnectionUiState.Step.ResolvingRoom)) } @@ -258,19 +276,31 @@ class AudioRoomViewModel( } } - /** Toggle the speaker-side mic mute. Cheap; the network keeps running. */ + /** + * Toggle the speaker-side mic mute. Cheap; the network keeps running. + * + * UI flips AFTER the suspending `broadcastHandle.setMuted(...)` returns + * so the indicator never claims muted while audio is still flowing + * (audit VM #7). We accept a small UI latency in exchange for an + * accurate state machine. + */ fun setMicMuted(muted: Boolean) { if (closed) return + val handle = broadcastHandle ?: return viewModelScope.launch { - broadcastHandle?.runCatching { setMuted(muted) } - } - _uiState.update { - val current = it.broadcast - if (current is BroadcastUiState.Broadcasting) { - it.copy(broadcast = current.copy(isMuted = muted)) - } else { - it - } + handle + .runCatching { setMuted(muted) } + .onSuccess { + if (closed) return@onSuccess + _uiState.update { + val current = it.broadcast + if (current is BroadcastUiState.Broadcasting) { + it.copy(broadcast = current.copy(isMuted = muted)) + } else { + it + } + } + } } } @@ -286,6 +316,11 @@ class AudioRoomViewModel( autoRetryJob?.cancel() autoRetryJob = null autoRetryAttempts = 0 + retryPending = false + // Forget the requested speaker set so a fresh connect() to a + // different room (or same room after a long pause) doesn't reuse + // a stale snapshot that may no longer be on stage. + requestedSpeakers = emptySet() teardownBroadcast(BroadcastUiState.Idle, finalCleanup = false) teardown(targetState = ConnectionUiState.Idle, finalCleanup = false) } @@ -383,24 +418,34 @@ class AudioRoomViewModel( * Auto-reconnect on listener Failed with capped exponential backoff. * The user can still tap Connect manually at any time; that resets the * retry counter via the `connect()` path. + * + * `retryPending` is the single source of truth — `Job.isActive` returns + * false the moment a coroutine starts running its body, which created a + * window where two scheduleAutoRetry calls could both pass the gate + * (audit VM #4). */ private fun scheduleAutoRetry() { if (closed) return - if (autoRetryJob?.isActive == true) return + if (retryPending) return if (autoRetryAttempts >= MAX_AUTO_RETRIES) return + retryPending = true val attempt = autoRetryAttempts autoRetryAttempts = attempt + 1 val backoffMs = minOf(MAX_RETRY_BACKOFF_MS, INITIAL_RETRY_BACKOFF_MS shl attempt) autoRetryJob = viewModelScope.launch { - delay(backoffMs) - if (closed) return@launch - if (_uiState.value.connection !is ConnectionUiState.Failed) return@launch - // Drop the previous listener cleanly before starting a new - // attempt. Reset state to Idle so the connect() guard passes. - teardown(targetState = ConnectionUiState.Idle) - connectInternal() + try { + delay(backoffMs) + if (closed) return@launch + if (_uiState.value.connection !is ConnectionUiState.Failed) return@launch + // Drop the previous listener cleanly before starting a new + // attempt. Reset state to Idle so the connect() guard passes. + teardown(targetState = ConnectionUiState.Idle) + connectInternal() + } finally { + retryPending = false + } } } @@ -469,17 +514,21 @@ class AudioRoomViewModel( } /** - * Stop the per-speaker player synchronously (releases the audio device - * immediately) and fire-and-forget the MoQ UNSUBSCRIBE on the VM scope. + * Stop the per-speaker player + fire-and-forget UNSUBSCRIBE on the VM + * scope. Both `AudioRoomPlayer.stop()` and `SubscribeHandle.unsubscribe()` + * are suspend, so we route them through one coroutine instead of two. */ private fun closeSubscription(slot: ActiveSubscription) { - val handle = slot.detach() + val (roomPlayer, handle) = slot.detach() speakingExpiryJobs.remove(slot.pubkey)?.cancel() if (_uiState.value.speakingNow.contains(slot.pubkey)) { _uiState.update { it.copy(speakingNow = (it.speakingNow - slot.pubkey).toPersistentSet()) } } - if (handle != null) { - viewModelScope.launch { runCatching { handle.unsubscribe() } } + if (roomPlayer != null || handle != null) { + viewModelScope.launch { + roomPlayer?.runCatching { stop() } + handle?.runCatching { unsubscribe() } + } } } @@ -532,22 +581,29 @@ class AudioRoomViewModel( connectJob = null stateObserverJob?.cancel() stateObserverJob = null - // Stop players synchronously — unsubscribe happens implicitly when - // the listener.close() below tears down the MoQ session. - activeSubscriptions.values.forEach { it.detach() } + // Detach + suspend-stop each player on the cleanup scope. The + // listener.close() below tears down the MoQ session and drops + // every active subscription, so we don't need to call + // unsubscribe() per-handle here — but we DO need to await each + // AudioRoomPlayer.stop() so the native MediaCodec / AudioTrack + // is released after its decode loop has unwound. + val scope = if (finalCleanup) cleanupScope else viewModelScope + val players = activeSubscriptions.values.map { it.detach().first } activeSubscriptions.clear() speakingExpiryJobs.values.forEach { it.cancel() } speakingExpiryJobs.clear() + if (players.isNotEmpty()) { + scope.launch { + players.forEach { p -> p?.runCatching { stop() } } + } + } val l = listener listener = null if (l != null) { // 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 + // before the QUIC transport drops. Same scope rule as the + // player stop above: viewModelScope when alive, cleanupScope + // for onCleared so the wire teardown survives. scope.launch { runCatching { l.close() } } } _uiState.update { @@ -615,18 +671,20 @@ class AudioRoomViewModel( } /** - * Stop the player + decoder synchronously and return the - * [SubscribeHandle] (if any) so the caller can fire-and-forget - * UNSUBSCRIBE on its own coroutine scope. + * Hand the player + handle back to the caller's coroutine scope — + * `AudioRoomPlayer.stop()` and `SubscribeHandle.unsubscribe()` are + * both suspend, and the caller has the right scope to await them + * (so native MediaCodec/AudioTrack release runs after the decode + * loop has unwound, per audit MoQ #11/#12). */ - fun detach(): SubscribeHandle? { + fun detach(): Pair { isPlaying = false - roomPlayer?.let { runCatching { it.stop() } } + val p = roomPlayer val h = handle roomPlayer = null handle = null player = null - return h + return p to h } companion object { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt index a0e920e10..ed474dcd1 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsSpeaker.kt @@ -199,13 +199,23 @@ class DefaultNestsSpeaker internal constructor( } override suspend fun close() { + // Take the active handle under `gate` (so a concurrent + // `startBroadcasting` can't observe a half-closed state), then + // release the lock before calling `handle.close()` and + // `session.close()` — both are long-running suspend operations + // (broadcaster.stop awaits cancelAndJoin; session.close fires + // SUBSCRIBE_DONE per attached subscriber and joins pumps). Holding + // the gate through them would block any other API call on this + // speaker for the entire teardown duration. + val handle: DefaultBroadcastHandle? gate.withLock { if (state.value is NestsSpeakerState.Closed) return - activeHandle?.let { runCatching { it.close() } } + handle = activeHandle activeHandle = null - runCatching { session.close() } mutableState.value = NestsSpeakerState.Closed } + handle?.runCatching { close() } + runCatching { session.close() } } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt index af98c46b0..d4cf38de7 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt @@ -24,6 +24,7 @@ import com.vitorpamplona.nestsclient.moq.MoqSession import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.launch /** @@ -129,11 +130,18 @@ class AudioRoomBroadcaster( * Stop the loop, release the mic, release the encoder, close the MoQ * publisher (which fires SUBSCRIBE_DONE to every attached subscriber). * Idempotent. + * + * Implementation note: we `cancelAndJoin` the loop before releasing + * the encoder and closing the publisher. Otherwise the loop's last + * `encoder.encode(...)` or `publisher.send(...)` could race + * `encoder.release()` / `publisher.close()` and produce orphan + * OBJECT_DATAGRAMs to subscribers that already received SUBSCRIBE_DONE, + * or use-after-release on the native MediaCodec. */ suspend fun stop() { if (stopped) return stopped = true - job?.cancel() + job?.cancelAndJoin() runCatching { capture.stop() } runCatching { encoder.release() } runCatching { publisher.close() } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayer.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayer.kt index 6601171fa..f74af9267 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayer.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomPlayer.kt @@ -24,6 +24,7 @@ import com.vitorpamplona.nestsclient.moq.MoqObject import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.launch @@ -101,11 +102,19 @@ class AudioRoomPlayer( } } - /** Stop playback, cancel the decode loop, release the decoder. Idempotent. */ - fun stop() { + /** + * Stop playback, cancel the decode loop, release the decoder. Idempotent. + * + * Suspending so callers can await the loop's exit before releasing + * native resources. Calling `decoder.release()` while another coroutine + * is mid-`decoder.decode(...)` is undefined behaviour for MediaCodec + * (and most native decoders); `cancelAndJoin` waits for the loop to + * unwind through its CancellationException path before we proceed. + */ + suspend fun stop() { if (stopped) return stopped = true - job?.cancel() + job?.cancelAndJoin() runCatching { player.stop() } runCatching { decoder.release() } } 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 ce43b6208..104e20add 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt @@ -24,6 +24,7 @@ import com.vitorpamplona.nestsclient.moq.MoqSession.Companion.client import com.vitorpamplona.nestsclient.moq.MoqSession.Companion.server import com.vitorpamplona.nestsclient.transport.WebTransportBidiStream import com.vitorpamplona.nestsclient.transport.WebTransportSession +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job @@ -356,13 +357,31 @@ class MoqSession private constructor( // ---------------------------------------------------------------- Pumps private fun startPumps() { + // If a pump exits unexpectedly (transport died, peer hung up), the + // session is no longer healthy — failing in-flight subscribe/announce + // waiters and flipping `closed=true` prevents new operations from + // hanging on a peer that will never reply. The control pump is the + // primary signal for transport death; the datagram pump exiting on + // its own is rare but treated the same way. controlPumpJob = pumpScope.launch { - runControlPump() + try { + runControlPump() + } catch (ce: CancellationException) { + throw ce + } catch (t: Throwable) { + runCatching { close(0, "control pump failed: ${t.message}") } + } } datagramPumpJob = pumpScope.launch { - runDatagramPump() + try { + runDatagramPump() + } catch (ce: CancellationException) { + throw ce + } catch (t: Throwable) { + runCatching { close(0, "datagram pump failed: ${t.message}") } + } } } @@ -424,16 +443,41 @@ class MoqSession private constructor( } is AnnounceError -> { - val deferred = + // Two cases: + // (a) ANNOUNCE_ERROR arrives BEFORE ANNOUNCE_OK — the peer + // refused our announce. The deferred is still pending; + // fail it and remove the announces[] entry we + // optimistically inserted in announce(). + // (b) ANNOUNCE_ERROR arrives AFTER ANNOUNCE_OK — the peer + // is revoking a previously-accepted announce + // (session-level kick, e.g. policy change). Mark the + // handle session-closed so further sends/openTracks + // short-circuit, but don't remove from announces[] + // since unannounce() is still the canonical removal + // path and inbound SUBSCRIBE must continue to see the + // namespace as "withdrawn" rather than missing. + val handle = stateMutex.withLock { - announces.remove(msg.namespace) - pendingAnnounces.remove(msg.namespace) + val pending = pendingAnnounces.remove(msg.namespace) + if (pending != null) { + announces.remove(msg.namespace) + pending.completeExceptionally( + MoqProtocolException( + "ANNOUNCE rejected: code=${msg.errorCode} reason=${msg.reasonPhrase}", + ), + ) + null + } else { + announces[msg.namespace] + } } - deferred?.completeExceptionally( - MoqProtocolException( - "ANNOUNCE rejected: code=${msg.errorCode} reason=${msg.reasonPhrase}", - ), - ) + handle?.let { existing -> + stateMutex.withLock { existing.markSessionClosedLocked() } + runCatching { + writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(msg.namespace))) } + } + stateMutex.withLock { announces.remove(msg.namespace) } + } } is Subscribe -> { @@ -573,8 +617,18 @@ class MoqSession private constructor( inboundSubscribers.clear() publisherSubscribers.clear() } - controlPumpJob?.cancel() - datagramPumpJob?.cancel() + // Cancel pumps then JOIN them, so any in-flight pump write + // (handleInboundSubscribe writing SUBSCRIBE_OK / SUBSCRIBE_ERROR, + // handleInboundUnsubscribe writing SUBSCRIBE_DONE) finishes its + // writeMutex.withLock { ... } critical section before we send FIN + // on the control stream. Joining a cancelled coroutine waits for + // its `finally`s to run, including the lock release. + val ctrlPump = controlPumpJob + val datagramPump = datagramPumpJob + ctrlPump?.cancel() + datagramPump?.cancel() + runCatching { ctrlPump?.join() } + runCatching { datagramPump?.join() } runCatching { writeMutex.withLock { controlStream.finish() } } runCatching { transport.close(code, reason) } } @@ -673,13 +727,24 @@ class MoqSession private constructor( sessionClosed = true toClose = tracks.values.toList() tracks.clear() - announces.remove(namespace) + // Note: keep `announces[namespace] = this` until UNANNOUNCE + // is on the wire. Inbound SUBSCRIBE during this window will + // see `sessionClosed=true` via publisherForLocked → null → + // we reply SUBSCRIBE_ERROR(TRACK_DOES_NOT_EXIST), which is + // accurate (no track to serve). Removing now would race + // with the control pump and leave a window where the peer + // thinks the namespace exists but our state has dropped it. } toClose.forEach { runCatching { it.close() } } - if (closed) return - runCatching { - writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(namespace))) } + if (!closed) { + runCatching { + writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(namespace))) } + } } + // Now safe to forget the namespace entirely — UNANNOUNCE is on + // the wire so any later inbound SUBSCRIBE was sent without + // knowing the namespace was withdrawn. + stateMutex.withLock { announces.remove(namespace) } } /** @@ -728,6 +793,16 @@ class MoqSession private constructor( runCatching { transport.sendDatagram(datagram) } .onSuccess { anySent = true } } + // Roll the objectId back if the entire fan-out failed (e.g. + // transport down). Audio-rooms NIP wants strictly contiguous + // object ids per group; a gap from a fully-failed send would + // trip strict subscribers. We roll back only if the next send + // hasn't already grabbed an id past us. + if (!anySent) { + stateMutex.withLock { + if (nextObjectId == objectId + 1) nextObjectId = objectId + } + } return anySent }