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 }