nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(nestsClient): fix race + leaks + Android encoder/decoder bugs from second audit

Browse Source 
A fresh audit caught a real race in the round-1 subscribe() rewrite plus
several distinct bugs in the Android MediaCodec layer.

**MoqLiteSession.subscribe() race regression** — the launched collector
that reads the SubscribeResponse and watches for peer FIN ran cleanup
(remove from subscriptionsBySubscribeId, close frames) before subscribe()
itself reached the post-await registration. If the publisher FIN'd
immediately after sending Ok, the collector exited first against an
empty map; subscribe() then registered the subscription, leaving frames
in the map with no live collector to ever close it on transport drop.
Consumer hung forever — exactly the failure mode round-1 fix #3 was
supposed to prevent. Fix: pre-register the subscription BEFORE launching
the collector. Drop unused `ok` field from ListenerSubscription.

**MoqLiteNestsSpeaker.startBroadcasting publisher leak** — if
session.publish() succeeded but broadcaster.start() then threw (mic
permission denied, AudioRecord allocation failure), the publisher was
never closed and stayed registered as session.activePublisher,
permanently blocking subsequent startBroadcasting calls.

**runCatching{suspend close} swallowing CancellationException** —
MoqLiteNestsSpeaker.close, MoqLiteBroadcastHandle.close,
MoqLiteNestsListener.close all wrapped suspending closes in
runCatching, breaking structured cancellation when the parent scope
cancelled teardown. Replaced with explicit cancel-rethrowing try/catch.

**MediaCodecOpusDecoder ArrayList<Short> boxing on every audio frame**
— at 50 fps × 960 samples × N speakers ≈ 48 000 boxed Short
allocations/sec/speaker on the audio hot path. Rewritten to write
directly into a pre-sized ShortArray via ShortBuffer.get(dst, off, len).

**MediaCodecOpusEncoder bugs**
- KEY_AAC_PROFILE = AACObjectLC was being set on an audio/opus
  encoder. Meaningless for Opus; stricter Codec2 stacks on Android
  13+ reject the configure() call with IllegalArgumentException and
  surface as DeviceUnavailable. Removed.
- The drain loop's INFO_OUTPUT_FORMAT_CHANGED branch had no progress
  guard. A buggy encoder re-emitting FORMAT_CHANGED without producing
  output would busy-spin against the 10 ms dequeue timeout. Now
  absorbed at most once per encode call. Same guard added to the
  decoder.

Adds regression test:
  - frames_flow_completes_when_peer_FINs_immediately_after_Ok

224 tests pass, 0 failures. Android target compiles clean.
This commit is contained in:
Claude
2026-05-01 18:28:41 +00:00
parent 9729f3a20c
commit 702885f4cb
6 changed files with 174 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt
+25 -11
View File
@@ -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<Short>(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<Short>, 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() {
nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusEncoder.kt
+13 -8
View File
@@ -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.
}
}
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt
+18 -2
View File
@@ -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
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt
+56 -15
View File
@@ -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)
}
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
+30 -17
View File
@@ -225,6 +225,29 @@ class MoqLiteSession internal constructor(
// on dead UDP paths.
val frames = Channel<MoqLiteFrame>(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST)
val responseDeferred = CompletableDeferred<MoqLiteCodec.SubscribeResponse>()
// 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<MoqLiteFrame>,
)
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt
+32
View File
@@ -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 {