fix(nests): NestPlayer keeps existing decoder when boundary factory throws
Audit-1: the publisher-boundary rebuild path released the old decoder
BEFORE asking the factory for a replacement (`runCatching {
decoder.release() }; decoder = factory()`). If `factory()` threw —
MediaCodec contention, audio policy denial mid-session, etc. — the
released decoder stayed assigned to the field and every subsequent
`decoder.decode(payload)` would throw `IllegalStateException` with
no recovery path. The room would silently fail for that subscription
until torn down.
Build the replacement first; only release the old one and swap on
success. On factory failure, log and keep the existing decoder
running. Cross-publisher Opus predictor state is wrong for one
group but at least audio keeps playing.
Audit-4+5: companion test cleanup —
- Drop the `byteArrayOf(0x0A) + it` / `0x0B` dead expressions in
the FakeOpusDecoder closures of the existing boundary-rebuild
test. They allocated a ByteArray and concatenated, then threw
the result away.
- Drop the local `IntBox` helper and use
`java.util.concurrent.atomic.AtomicInteger` like the rest of
the codebase does (`MoqLiteSession`, `MoqLiteNestsListener`).
Same semantics, no new vocabulary.
New test pins the dangling-field fix:
`publisher_boundary_keeps_old_decoder_when_factory_throws` flows
two MoqObjects with different trackAliases through a NestPlayer
whose factory always throws; asserts both frames decode through
the original decoder and the original decoder is released exactly
once on `stop()`.
https://claude.ai/code/session_014JfZJHSTvyYYWJbC9VbB47
This commit is contained in:
Claude
+27
-4
@@ -216,13 +216,36 @@ class NestPlayer(
|
||||
// side) or cliff-detector recycle (listener side).
|
||||
// The prior-trackAlias guard avoids a spurious
|
||||
// rebuild on the very first frame.
|
||||
//
|
||||
// Build the replacement BEFORE releasing the old
|
||||
// decoder so a factory failure (e.g. MediaCodec
|
||||
// contention, audio policy denial mid-session)
|
||||
// doesn't leave the field referencing a
|
||||
// released codec — every subsequent decode
|
||||
// would then throw `IllegalStateException` with
|
||||
// no recovery path. On factory failure, log
|
||||
// and keep using the existing decoder; cross-
|
||||
// publisher predictor state is wrong but at
|
||||
// least audio keeps playing.
|
||||
val factory = decoderFactory
|
||||
if (factory != null && lastTrackAlias != null && obj.trackAlias != lastTrackAlias) {
|
||||
com.vitorpamplona.quartz.utils.Log.d("NestPlay") {
|
||||
"NestPlayer publisher boundary: trackAlias $lastTrackAlias → ${obj.trackAlias}; rebuilding decoder"
|
||||
val replacement =
|
||||
runCatching { factory() }
|
||||
.onFailure { t ->
|
||||
if (t is CancellationException) throw t
|
||||
com.vitorpamplona.quartz.utils.Log.w("NestPlay") {
|
||||
"NestPlayer decoder factory threw on trackAlias " +
|
||||
"$lastTrackAlias → ${obj.trackAlias}; keeping old decoder " +
|
||||
"(${t::class.simpleName}: ${t.message})"
|
||||
}
|
||||
}.getOrNull()
|
||||
if (replacement != null) {
|
||||
com.vitorpamplona.quartz.utils.Log.d("NestPlay") {
|
||||
"NestPlayer publisher boundary: trackAlias $lastTrackAlias → ${obj.trackAlias}; rebuilding decoder"
|
||||
}
|
||||
runCatching { decoder.release() }
|
||||
decoder = replacement
|
||||
}
|
||||
runCatching { decoder.release() }
|
||||
decoder = factory()
|
||||
}
|
||||
lastTrackAlias = obj.trackAlias
|
||||
val pcm =
|
||||
|
||||
+46
-26
@@ -25,6 +25,7 @@ import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.flowOf
|
||||
import kotlinx.coroutines.flow.receiveAsFlow
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertContentEquals
|
||||
import kotlin.test.assertEquals
|
||||
@@ -346,17 +347,9 @@ class NestPlayerTest {
|
||||
// Two distinct decoders so we can prove the factory was
|
||||
// invoked. After the trackAlias change, frames should
|
||||
// route through `decoderB`, NOT `decoderA`.
|
||||
val decoderA =
|
||||
FakeOpusDecoder {
|
||||
byteArrayOf(0x0A) + it
|
||||
ShortArray(it.size) { _ -> 0xAA.toShort() }
|
||||
}
|
||||
val decoderB =
|
||||
FakeOpusDecoder {
|
||||
byteArrayOf(0x0B) + it
|
||||
ShortArray(it.size) { _ -> 0xBB.toShort() }
|
||||
}
|
||||
val factoryCallCount = atomicIntZero()
|
||||
val decoderA = FakeOpusDecoder { ShortArray(it.size) { _ -> 0xAA.toShort() } }
|
||||
val decoderB = FakeOpusDecoder { ShortArray(it.size) { _ -> 0xBB.toShort() } }
|
||||
val factoryCallCount = AtomicInteger(0)
|
||||
val factory: () -> OpusDecoder = {
|
||||
if (factoryCallCount.getAndIncrement() == 0) decoderA else decoderB
|
||||
}
|
||||
@@ -384,13 +377,54 @@ class NestPlayerTest {
|
||||
sut.play(objects)
|
||||
testScheduler.advanceUntilIdle()
|
||||
|
||||
assertEquals(2, factoryCallCount.value, "factory invoked twice: initial + boundary")
|
||||
assertEquals(2, factoryCallCount.get(), "factory invoked twice: initial + boundary")
|
||||
assertEquals(1, decoderA.releaseCount, "decoderA released on the boundary")
|
||||
assertEquals(0, decoderB.releaseCount, "decoderB still alive (released on stop)")
|
||||
sut.stop()
|
||||
assertEquals(1, decoderB.releaseCount, "decoderB released on stop")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun publisher_boundary_keeps_old_decoder_when_factory_throws() =
|
||||
runTest {
|
||||
// The decoder field MUST NOT be left referencing a released
|
||||
// decoder if the factory throws — every subsequent decode
|
||||
// would otherwise fail with `IllegalStateException` and the
|
||||
// subscription would be permanently dead.
|
||||
val decoderA = FakeOpusDecoder { ShortArray(it.size) { _ -> 0xAA.toShort() } }
|
||||
val factoryFailures = AtomicInteger(0)
|
||||
val factory: () -> OpusDecoder = {
|
||||
factoryFailures.incrementAndGet()
|
||||
throw IllegalStateException("synthetic factory failure")
|
||||
}
|
||||
val player = FakeAudioPlayer()
|
||||
|
||||
val objects =
|
||||
flowOf(
|
||||
moqObject(byteArrayOf(0x01), trackAlias = 7L),
|
||||
// Boundary: factory throws → decoderA must stay alive
|
||||
// and decode the next frame normally.
|
||||
moqObject(byteArrayOf(0x02), trackAlias = 8L),
|
||||
)
|
||||
|
||||
val sut =
|
||||
NestPlayer(
|
||||
initialDecoder = decoderA,
|
||||
player = player,
|
||||
scope = this,
|
||||
decoderFactory = factory,
|
||||
)
|
||||
sut.play(objects)
|
||||
testScheduler.advanceUntilIdle()
|
||||
|
||||
assertEquals(1, factoryFailures.get(), "factory invoked once on the boundary")
|
||||
// decoderA still alive → both frames decoded through it.
|
||||
assertEquals(2, player.queued.size)
|
||||
assertEquals(0, decoderA.releaseCount, "decoderA NOT released on factory failure")
|
||||
sut.stop()
|
||||
assertEquals(1, decoderA.releaseCount, "decoderA released exactly once on stop")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun publisher_boundary_no_op_when_factory_is_null() =
|
||||
runTest {
|
||||
@@ -427,20 +461,6 @@ class NestPlayerTest {
|
||||
payload = payload,
|
||||
)
|
||||
|
||||
/**
|
||||
* Tiny stand-in for AtomicInteger that's available in commonMain
|
||||
* (kotlin.test scope). Used by the boundary-rebuild test to count
|
||||
* factory invocations across the test scope's coroutine
|
||||
* dispatcher.
|
||||
*/
|
||||
private class IntBox {
|
||||
var value: Int = 0
|
||||
|
||||
fun getAndIncrement(): Int = value++
|
||||
}
|
||||
|
||||
private fun atomicIntZero(): IntBox = IntBox()
|
||||
|
||||
private fun byteToShorts(b: ByteArray): ShortArray = ShortArray(b.size) { b[it].toShort() }
|
||||
|
||||
private class FakeOpusDecoder(
|
||||
|
||||
Reference in New Issue
Block a user