From ed793e8eb38b11dbd676e2ea9ac6a61a524ac599 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 12:32:22 +0000 Subject: [PATCH] =?UTF-8?q?fix(audio-rooms):=20round-2=20audit=20=E2=80=94?= =?UTF-8?q?=20pump=20self-join=20(CRITICAL)=20+=205=20other=20findings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 audit (3 parallel agents reviewing every change since the previous follow-up commit) caught one CRITICAL regression and several HIGH/MED items. Most round-1 fixes verified clean. CRITICAL fix (audit round-2 MoQ #1): - Pump exception handlers added in the previous commit call `close()` from inside the failing pump's own coroutine. `close()` now does `controlPumpJob?.join()` to drain in-flight writes — but the Job we try to join is the very Job we're inside, so `join()` suspends forever (lambda can't finish until close returns; close can't return until lambda finishes). `runCatching` doesn't help — `join()` doesn't throw, it suspends. This deadlocks the entire session whenever a pump fails. Fix: skip the join when the current coroutine IS the job we're joining. `currentCoroutineContext()[Job]` identifies the caller; we compare and bypass. HIGH fixes: - MoQ #2 (regression): concurrent unannounce() + post-OK AnnounceError handler could both write UNANNOUNCE on the wire (some relays disconnect on UNANNOUNCE for an unknown namespace). Fix: `AnnounceHandleImpl.unannounceWritten: Boolean` flag, set under stateMutex by whichever writer goes first; the other path skips. - VM #3 (new): `connect()` overwrote `listener` if invoked from a Failed-with-stale-listener state, leaking the previous MoQ session. Fix: call `teardown(targetState=Idle, finalCleanup=false)` before launching the new attempt when listener or stateObserverJob is still alive. - VM #7 (new): `openSubscription` allocated decoder + player via the factories, then attached them to the slot. If the VM scope was cancelled between `decoderFactory()` and `slot.attach(...)`, the native MediaCodec / AudioTrack leaked because nothing was tracking them yet. Fix: nest a try/catch that releases both on any throw (including `CancellationException`) before re-throwing. MED fixes: - MoQ #7 (new): `capture.start()` could throw before `job` was assigned, leaving the broadcaster in a half-started state where future `start()` calls would re-pass the guards and double-start the mic. Fix: try/catch around capture.start; on throw, set `stopped = true` + run capture.stop and propagate. - MoQ #8 (new): `stopped` was read across threads (setMuted from caller, stop from anywhere) without a `@Volatile` barrier. Visibility hazard. Fix: `@Volatile private var stopped`. - Android #12 (new): after the user granted RECORD_AUDIO via the Settings deep-link, `permissionDenied` stayed `true` because the launcher callback never fired — the warning + Open-settings button remained visible until the user tapped Talk again. Fix: derive `showDenialWarning` from `permissionDenied AND ContextCompat.checkSelfPermission(...) != GRANTED`. Re-checks every recomposition (including post-Settings return). Round-1 fixes verified clean by this audit: - pending-deferred completeExceptionally on close - SubscribeDoneStatus codes (UNSUBSCRIBED=0x00, TRACK_ENDED=0x03) - suspend `stop()` conversions on broadcaster + player - gate release before NestsSpeaker teardown chain - unannounce() ordering on thrown wire-write - SharedFlow PIP signal (rapid double-tap behavior is correct) - RECEIVER_NOT_EXPORTED gate (constant 4 doesn't collide; round-1 collision claim was incorrect) - onUserLeaveHint guards (PIP from lobby, no PIP support) - foreground service `startForeground`-always-first contract (`Result.onFailure` is `inline`, the `return` IS a non-local return from `onStartCommand` — verified) - 4-hour wake-lock cap - AudioRoomBridge.clear() in AccountViewModel.onCleared Still deferred: - VM #6 (round-1 carryover): unsynchronized speakingExpiryJobs map. Cross-thread mutation under contention. Needs ConcurrentHashMap or Dispatchers.Main.immediate marshalling. - VM #10 (round-2 new): brief two-QUIC-session overlap during rapid disconnect→connect. - Android #5 (round-2 new): same-room re-entry from MainActivity while in PIP doesn't auto-exit PIP. - MoQ #1, #8 (round-1 carryover): wire-format draft pinning. Still blocked on M4 manual interop input. Verified: spotlessApply clean; :commons:jvmTest, :nestsClient:jvmTest (80 tests), :amethyst:compilePlayDebugKotlin all green. --- .../audiorooms/room/AudioRoomFullScreen.kt | 18 +++++- .../commons/viewmodels/AudioRoomViewModel.kt | 59 +++++++++++++------ .../nestsclient/audio/AudioRoomBroadcaster.kt | 20 ++++++- .../nestsclient/moq/MoqSession.kt | 56 +++++++++++++++--- 4 files changed, 123 insertions(+), 30 deletions(-) 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 c3fa9378a..2fa567134 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 @@ -228,8 +228,22 @@ private fun TalkRow( var permissionDenied by rememberSaveable { mutableStateOf(false) } val permissionLauncher = rememberLauncherForActivityResult(ActivityResultContracts.RequestPermission()) { granted -> - if (granted) viewModel.startBroadcast(speakerPubkeyHex) else permissionDenied = true + if (granted) { + permissionDenied = false + viewModel.startBroadcast(speakerPubkeyHex) + } else { + permissionDenied = true + } } + // If the user grants RECORD_AUDIO via the system Settings deep-link + // and returns to the activity, the permissionLauncher callback never + // fires and `permissionDenied` would otherwise stay true. Recompute + // every time the permission state could have changed (audit round-2 + // Android #12). + val showDenialWarning = + permissionDenied && + ContextCompat.checkSelfPermission(context, Manifest.permission.RECORD_AUDIO) != + PackageManager.PERMISSION_GRANTED Row( modifier = Modifier.fillMaxWidth().padding(top = 8.dp), @@ -252,7 +266,7 @@ private fun TalkRow( }) { Text(stringRes(R.string.audio_room_talk)) } - if (permissionDenied) { + if (showDenialWarning) { Text( text = stringRes(R.string.audio_room_record_permission_required), style = MaterialTheme.typography.bodySmall, 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 d30614e3b..f8847ee8a 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 @@ -164,11 +164,16 @@ class AudioRoomViewModel( 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 + // If a stale listener is still set (e.g. arrived in Failed and the + // user is manually retrying before the auto-retry fires, or + // entering from a server-Closed-but-not-yet-disconnected state), + // tear it down so the new connect doesn't leave the old MoQ + // session open and unowned (audit round-2 VM #3). teardown() + // also cancels stateObserverJob so its delayed emissions can't + // clobber the fresh Connecting UI. + if (listener != null || stateObserverJob != null) { + teardown(targetState = ConnectionUiState.Idle, finalCleanup = false) + } _uiState.update { it.copy(connection = ConnectionUiState.Connecting(ConnectionUiState.Step.ResolvingRoom)) } @@ -549,19 +554,39 @@ class AudioRoomViewModel( viewModelScope.launch { runCatching { handle.unsubscribe() } } return } + // Allocate native resources (MediaCodec decoder + AudioTrack + // player on Android). Both are heavy and leaky if dropped on + // the floor — wrap them in a nested try so any cancellation + // or throw between here and slot.attach releases them + // (audit round-2 VM #7). val decoder = decoderFactory() - val player = playerFactory() - val isMuted = _uiState.value.isMuted - val roomPlayer = AudioRoomPlayer(decoder, player, viewModelScope) - // Apply current mute state before play() opens the device so the - // first frame respects it. - player.setMutedSafe(isMuted) - // Tap the object flow to drive the speaking-now indicator before - // the decoder consumes it. - val instrumented = handle.objects.onEach { onSpeakerActivity(pubkey) } - roomPlayer.play(instrumented, onError = { /* swallow per-packet decoder errors */ }) - slot.attach(handle, roomPlayer, player) - publishActiveSpeakers() + val player = + try { + playerFactory() + } catch (t: Throwable) { + runCatching { decoder.release() } + throw t + } + try { + val isMuted = _uiState.value.isMuted + val roomPlayer = AudioRoomPlayer(decoder, player, viewModelScope) + // Apply current mute state before play() opens the device so the + // first frame respects it. + player.setMutedSafe(isMuted) + // Tap the object flow to drive the speaking-now indicator before + // the decoder consumes it. + val instrumented = handle.objects.onEach { onSpeakerActivity(pubkey) } + roomPlayer.play(instrumented, onError = { /* swallow per-packet decoder errors */ }) + slot.attach(handle, roomPlayer, player) + publishActiveSpeakers() + } catch (t: Throwable) { + // Either CancellationException (scope cancelled mid-construction) + // or an unexpected throw — release the half-built pipeline and + // re-throw so the outer catch handles slot rollback. + runCatching { player.stop() } + runCatching { decoder.release() } + throw t + } } catch (ce: CancellationException) { throw ce } catch (t: Throwable) { 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 d4cf38de7..45cddbd77 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioRoomBroadcaster.kt @@ -53,19 +53,33 @@ class AudioRoomBroadcaster( private val scope: CoroutineScope, ) { private var job: Job? = null - private var stopped = false + + @Volatile private var stopped: Boolean = false @Volatile private var muted: Boolean = false /** * Start capturing + encoding + publishing in the background. Returns - * immediately. Calling twice is an error. + * immediately. Calling twice is an error. If [AudioCapture.start] + * throws (mic device unavailable, etc.), the broadcaster is left in + * a stopped state and the exception propagates so the caller can + * surface it to the user. */ fun start(onError: (AudioException) -> Unit = { /* swallow */ }) { check(!stopped) { "AudioRoomBroadcaster already stopped" } check(job == null) { "AudioRoomBroadcaster.start already called" } - capture.start() + // Audit round-2 MoQ #7: capture.start() can throw before we have + // a job. Mark stopped + propagate so a half-started capture isn't + // left holding the mic and the broadcaster instance can't be + // accidentally re-started. + try { + capture.start() + } catch (t: Throwable) { + stopped = true + runCatching { capture.stop() } + throw t + } job = scope.launch { try { 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 104e20add..9273cf7e8 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt @@ -30,6 +30,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.receiveAsFlow @@ -472,9 +473,23 @@ class MoqSession private constructor( } } handle?.let { existing -> - stateMutex.withLock { existing.markSessionClosedLocked() } - runCatching { - writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(msg.namespace))) } + val shouldWrite = + stateMutex.withLock { + existing.markSessionClosedLocked() + // Same compare-and-set as unannounce() to avoid + // a duplicate UNANNOUNCE on the wire when a + // concurrent unannounce() is mid-flight. + if (existing.unannounceWritten) { + false + } else { + existing.unannounceWritten = true + true + } + } + if (shouldWrite) { + runCatching { + writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(msg.namespace))) } + } } stateMutex.withLock { announces.remove(msg.namespace) } } @@ -621,14 +636,21 @@ class MoqSession private constructor( // (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. + // on the control stream. + // + // Self-join guard: when close() is called from a pump's own + // exception handler (the try/catch in startPumps), `coroutineContext[Job]` + // is the very Job we're trying to join. Joining ourselves + // suspends forever — the lambda can't finish until close() returns, + // and close() can't return until the lambda finishes. Skip the + // join in that case (round-2 audit MoQ #1). val ctrlPump = controlPumpJob val datagramPump = datagramPumpJob ctrlPump?.cancel() datagramPump?.cancel() - runCatching { ctrlPump?.join() } - runCatching { datagramPump?.join() } + val self = currentCoroutineContext()[Job] + runCatching { if (ctrlPump != null && ctrlPump !== self) ctrlPump.join() } + runCatching { if (datagramPump != null && datagramPump !== self) datagramPump.join() } runCatching { writeMutex.withLock { controlStream.finish() } } runCatching { transport.close(code, reason) } } @@ -704,6 +726,11 @@ class MoqSession private constructor( internal val tracks = HashMap() internal var sessionClosed = false + // True once UNANNOUNCE has been written on the wire, so a + // concurrent unannounce() + post-OK AnnounceError handler don't + // both emit it (audit round-2 MoQ #2). Read/written under stateMutex. + internal var unannounceWritten = false + override suspend fun openTrack(name: ByteArray): TrackPublisher { val key = TrackKey(name) val publisher = TrackPublisherImpl(name) @@ -736,7 +763,20 @@ class MoqSession private constructor( // thinks the namespace exists but our state has dropped it. } toClose.forEach { runCatching { it.close() } } - if (!closed) { + // Race guard: a post-OK AnnounceError handler may have already + // written UNANNOUNCE for this namespace (audit round-2 MoQ #2). + // Compare-and-set under stateMutex so only one writer fires + // the wire frame. + val shouldWrite = + stateMutex.withLock { + if (unannounceWritten) { + false + } else { + unannounceWritten = true + true + } + } + if (shouldWrite && !closed) { runCatching { writeMutex.withLock { controlStream.write(MoqCodec.encode(Unannounce(namespace))) } }