diff --git a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt index 23c9e3bca..24e08e2cc 100644 --- a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt +++ b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt @@ -74,33 +74,47 @@ class MediaCodecOpusDecoder : OpusDecoder { // Drain whatever output is ready right now. A single Opus packet // typically yields exactly one output buffer, but on some devices the // first call returns INFO_OUTPUT_FORMAT_CHANGED before the PCM frame. - val collected = ArrayList(AudioFormat.FRAME_SIZE_SAMPLES) - while (true) { + // + // Write the decoded samples directly into a pre-sized ShortArray + // via ShortBuffer.get(dst, off, len) — the previous shape went via + // ArrayList, which boxed every PCM sample (one heap object + // per Short × 960 samples × 50 fps × N speakers ≈ 48 000 + // allocations/sec/speaker on the audio hot path). + val out = ShortArray(AudioFormat.FRAME_SIZE_SAMPLES) + var outPos = 0 + var formatChangeAbsorbed = false + drain@ while (true) { val outputIndex = codec.dequeueOutputBuffer(bufferInfo, DEQUEUE_TIMEOUT_US) when { outputIndex >= 0 -> { val outputBuffer = codec.getOutputBuffer(outputIndex) - ?: continue + ?: continue@drain if (bufferInfo.size > 0) { outputBuffer.position(bufferInfo.offset) outputBuffer.limit(bufferInfo.offset + bufferInfo.size) val shorts = outputBuffer.order(ByteOrder.nativeOrder()).asShortBuffer() - val tmp = ShortArray(shorts.remaining()) - shorts.get(tmp) - for (s in tmp) collected.add(s) + val toCopy = minOf(shorts.remaining(), out.size - outPos) + if (toCopy > 0) { + shorts.get(out, outPos, toCopy) + outPos += toCopy + } } codec.releaseOutputBuffer(outputIndex, false) if (bufferInfo.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) break // No more buffered output for this packet. if (bufferInfo.size == 0) break - if (collected.size >= AudioFormat.FRAME_SIZE_SAMPLES) break + if (outPos >= out.size) break } outputIndex == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED -> { - // The format change carries no audio data; loop to read - // the actual PCM frame. - continue + // The format change carries no audio data; absorb it + // once and loop to read the actual PCM frame. Buggy + // decoders that re-fire format-change without + // producing output would otherwise busy-spin. + if (formatChangeAbsorbed) break + formatChangeAbsorbed = true + continue@drain } outputIndex == MediaCodec.INFO_TRY_AGAIN_LATER -> { @@ -112,7 +126,7 @@ class MediaCodecOpusDecoder : OpusDecoder { } } } - return ShortArray(collected.size) { collected[it] } + return if (outPos == out.size) out else out.copyOf(outPos) } override fun release() { diff --git a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusEncoder.kt b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusEncoder.kt index 9c406b2ed..5de9f3911 100644 --- a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusEncoder.kt +++ b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusEncoder.kt @@ -21,7 +21,6 @@ package com.vitorpamplona.nestsclient.audio import android.media.MediaCodec -import android.media.MediaCodecInfo import android.media.MediaFormat import java.nio.ByteOrder @@ -76,7 +75,12 @@ class MediaCodecOpusEncoder( presentationTimeUs += FRAME_DURATION_US // One PCM frame produces one Opus packet (sometimes after one warmup - // round). Drain the output queue once. + // round). Drain the output queue once. The format-change signal + // fires at most once on encoder startup; absorb it then re-poll, + // but never loop on it (some buggy encoders re-emit FORMAT_CHANGED + // without producing output, which would otherwise busy-spin at + // 100 Hz against the 10 ms dequeue timeout). + var formatChangeAbsorbed = false while (true) { val outputIndex = codec.dequeueOutputBuffer(bufferInfo, DEQUEUE_TIMEOUT_US) when { @@ -91,6 +95,8 @@ class MediaCodecOpusEncoder( } outputIndex == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED -> { + if (formatChangeAbsorbed) return ByteArray(0) + formatChangeAbsorbed = true continue } @@ -128,12 +134,11 @@ class MediaCodecOpusEncoder( ).apply { setInteger(MediaFormat.KEY_BIT_RATE, bitrate) setInteger(MediaFormat.KEY_PCM_ENCODING, android.media.AudioFormat.ENCODING_PCM_16BIT) - // Encoder-side AAC/Opus profile selection: SignalingDelaySamples - // is implicit; nothing else required for mono speech. - setInteger( - MediaFormat.KEY_AAC_PROFILE, - MediaCodecInfo.CodecProfileLevel.AACObjectLC, - ) + // KEY_AAC_PROFILE on an audio/opus encoder is meaningless + // (Opus has no AAC profile); historically it was silently + // ignored, but stricter Codec2 stacks on Android 13+ + // reject the configure() call with IllegalArgumentException + // and surface as DeviceUnavailable. } } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt index 670a114ea..5986ab11c 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt @@ -146,13 +146,29 @@ class MoqLiteNestsListener internal constructor( ) } } finally { - runCatching { handle.close() } + // The flow's `finally` runs on both normal completion + // and cancellation — let cancel propagate after closing + // the announce handle. + try { + handle.close() + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + // Best-effort. + } } } override suspend fun close() { if (state.value is NestsListenerState.Closed) return - runCatching { session.close() } + try { + session.close() + } catch (ce: kotlinx.coroutines.CancellationException) { + mutableState.value = NestsListenerState.Closed + throw ce + } catch (_: Throwable) { + // Best-effort. + } mutableState.value = NestsListenerState.Closed } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt index dfdf5836b..2b00d24ed 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt @@ -76,21 +76,27 @@ class MoqLiteNestsSpeaker internal constructor( // Per the audio-rooms NIP draft + JS reference // (`@moq/publish/screen-B680RFft.js:5641`), publishers // claim a broadcast suffix equal to their pubkey hex. - val publisher = + val publisher = session.publish(broadcastSuffix = speakerPubkeyHex) + // From here on, the publisher is registered in the session + // and has a live announce/subscribe path. Anything that + // throws before we hand a working handle back to the caller + // must close the publisher to avoid leaking it inside the + // session's `activePublisher` slot — otherwise a subsequent + // startBroadcasting fails with "already publishing" and the + // publisher's announce state never tears down. + val broadcaster = try { - session.publish(broadcastSuffix = speakerPubkeyHex) + NestMoqLiteBroadcaster( + capture = captureFactory(), + encoder = encoderFactory(), + publisher = publisher, + scope = scope, + framesPerGroup = framesPerGroup, + ).also { it.start() } } catch (t: Throwable) { + runCatching { publisher.close() } throw t } - val broadcaster = - NestMoqLiteBroadcaster( - capture = captureFactory(), - encoder = encoderFactory(), - publisher = publisher, - scope = scope, - framesPerGroup = framesPerGroup, - ) - broadcaster.start() mutableState.value = NestsSpeakerState.Broadcasting( room = current.room, @@ -141,8 +147,25 @@ class MoqLiteNestsSpeaker internal constructor( activeHandle = null mutableState.value = NestsSpeakerState.Closed } - handle?.runCatching { close() } - runCatching { session.close() } + // Don't `runCatching { handle.close() }` — that swallows + // CancellationException too, breaking structured cancellation + // when the parent scope is cancelling teardown. + if (handle != null) { + try { + handle.close() + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + // Best-effort — already cleared activeHandle. + } + } + try { + session.close() + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + // Best-effort. + } } } @@ -167,11 +190,29 @@ internal class MoqLiteBroadcastHandle( override suspend fun close() { if (closed) return closed = true - runCatching { broadcaster.stop() } + try { + broadcaster.stop() + } catch (ce: kotlinx.coroutines.CancellationException) { + // Even on cancel, run the rest of cleanup before rethrowing + // — broadcaster.stop already cancels its own job, so the + // mic + encoder + publisher are owed their close paths. + runCatching { publisher.close() } + parent.broadcastClosed(this) + throw ce + } catch (_: Throwable) { + // Best-effort; fall through to the defensive publisher.close. + } // broadcaster.stop() already calls publisher.close(); call again // defensively to make this method idempotent against partial // failures on the broadcaster.stop path. - runCatching { publisher.close() } + try { + publisher.close() + } catch (ce: kotlinx.coroutines.CancellationException) { + parent.broadcastClosed(this) + throw ce + } catch (_: Throwable) { + // Best-effort. + } parent.broadcastClosed(this) } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index f56a26824..20f216f0b 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -225,6 +225,29 @@ class MoqLiteSession internal constructor( // on dead UDP paths. val frames = Channel(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST) val responseDeferred = CompletableDeferred() + + // Pre-register the subscription BEFORE launching the collector. + // Otherwise: if the publisher FINs the bidi immediately after + // sending Ok, the collector's exit-cleanup races subscribe()'s + // post-await registration — collector exits, runs `remove(id)` + // against an empty map, no-ops; subscribe() then inserts; the + // frames channel is now in the map with no live collector to + // close it on transport tear-down. Consumer hangs forever. + // Pre-registering means the collector's idempotent + // remove+close cleanup always finds and closes the frames + // channel, regardless of timing. + val sub = + ListenerSubscription( + id = id, + request = request, + bidi = bidi, + frames = frames, + ) + state.withLock { + subscriptionsBySubscribeId[id] = sub + if (groupPump == null) groupPump = scope.launch { pumpUniStreams() } + } + scope.launch { val responseBuffer = MoqLiteFrameBuffer() var responseParsed = false @@ -257,10 +280,9 @@ class MoqLiteSession internal constructor( MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id"), ) } - // Close any frames channel registered for this id. The - // subscription may not be registered (response was Dropped - // or arrival raced cleanup) — in that case the snapshot - // returns null and we no-op. + // Idempotent: if subscribe() unwound on a Dropped response + // (or any throw from await), it already removed the + // subscription before throwing. Either way: remove + close. val removed = state.withLock { subscriptionsBySubscribeId.remove(id) } removed?.frames?.close() } @@ -269,11 +291,15 @@ class MoqLiteSession internal constructor( try { responseDeferred.await() } catch (t: Throwable) { + state.withLock { subscriptionsBySubscribeId.remove(id) } + frames.close() runCatching { bidi.finish() } throw t } when (resp) { is MoqLiteCodec.SubscribeResponse.Dropped -> { + state.withLock { subscriptionsBySubscribeId.remove(id) } + frames.close() runCatching { bidi.finish() } throw MoqLiteSubscribeException( "publisher rejected subscribe id=$id: errorCode=${resp.drop.errorCode} " + @@ -282,18 +308,6 @@ class MoqLiteSession internal constructor( } is MoqLiteCodec.SubscribeResponse.Ok -> { - val sub = - ListenerSubscription( - id = id, - request = request, - ok = resp.ok, - bidi = bidi, - frames = frames, - ) - state.withLock { - subscriptionsBySubscribeId[id] = sub - if (groupPump == null) groupPump = scope.launch { pumpUniStreams() } - } return MoqLiteSubscribeHandle( id = id, ok = resp.ok, @@ -707,7 +721,6 @@ class MoqLiteSession internal constructor( private class ListenerSubscription( val id: Long, val request: MoqLiteSubscribe, - val ok: MoqLiteSubscribeOk, val bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream, val frames: Channel, ) diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt index dcf3174ec..a258b8527 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt @@ -413,6 +413,38 @@ class MoqLiteSessionTest { session.close() } + @Test + fun frames_flow_completes_when_peer_FINs_immediately_after_Ok() = + runBlocking { + // Race regression: an earlier shape of subscribe() pre-registered + // the subscription AFTER awaiting the Ok response. If the peer + // FIN'd between sending Ok and subscribe() resuming, the + // collector's exit-cleanup ran first against an empty map, and + // subscribe() then inserted the subscription — leaving the + // frames channel registered with no live collector to ever close + // it. This test forces the order by writing Ok and FIN before + // the listener has a chance to consume them. + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + val peerJob = + async { + val (bidi, _) = nextSubscribeBidi(serverSide) + bidi.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L))) + bidi.finish() + } + + val handle = session.subscribe("speakerZ", "audio/data") + withTimeout(2_000) { peerJob.await() } + + // frames flow must complete naturally regardless of whether + // the FIN landed before or after subscribe()'s post-await + // registration. + withTimeout(2_000) { handle.frames.toList() } + + session.close() + } + @Test fun unsubscribe_FINs_the_subscribe_bidi() = runBlocking {