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

fix(nests): T13 reset Opus decoder on publisher boundary in NestPlayer

Browse Source 
The reissuing-subscribe wrapper splices fresh-publisher frames into the
same `SharedFlow` that NestPlayer consumes, so across a JWT-refresh
hot-swap (speaker side) or a cliff-detector recycle (listener side)
the decoder receives a discontinuous frame stream while keeping its
Opus predictor state. Result: an audible warble at every publisher
cycle. Single-stream tests don't catch it because they never present
two distinct publishers.

Detect the boundary via `MoqObject.trackAlias` — the underlying
`MoqLiteSession.subscribe` assigns a fresh subscribeId per SUBSCRIBE,
which the listener wrapper surfaces verbatim as `trackAlias` on every
emitted object. A change between consecutive objects = new publisher.

Add an optional `decoderFactory: (() -> OpusDecoder)?` to NestPlayer.
When non-null, NestPlayer tracks `lastTrackAlias` and on a change
releases the current decoder and rebuilds via the factory. The
default `null` preserves the legacy single-decoder behaviour so
existing tests / callers that don't care about boundaries stand
unchanged.

NestViewModel.openSubscription wires the factory: a closure capturing
the catalog-derived `channelCount` so a rebuild reuses the SAME
channel layout — without that capture, a rebuild after a stereo-
publisher cycle would default to mono and silently downmix.

Two new NestPlayerTest cases pin the behaviour:
  - `publisher_boundary_rebuilds_decoder_when_factory_provided`:
    factory invoked twice (initial + boundary), first decoder
    released on boundary, second released on stop.
  - `publisher_boundary_no_op_when_factory_is_null`: legacy path
    holds onto the same decoder across trackAlias changes (unchanged
    semantics).

NestPlayer's first constructor parameter is now `initialDecoder`
(was `decoder`); positional-arg call sites are unchanged but the
named-arg call sites in the test suite are updated accordingly.

https://claude.ai/code/session_014JfZJHSTvyYYWJbC9VbB47
This commit is contained in:
Claude
2026-05-06 18:00:52 +00:00
parent be4e0b9f98
commit 4714e3c723
3 changed files with 175 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt
+19 -2
View File
@@ -1177,7 +1177,16 @@ class NestViewModel(
// 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(channelCount)
// Per-subscription decoder factory closure: captures the
// catalog-derived [channelCount] so a publisher-boundary
// decoder rebuild (see [NestPlayer]'s `decoderFactory`
// kdoc) reuses the SAME channel layout — without it, a
// rebuild after a cliff-recycle would default to mono and
// a stereo publisher would silently downmix on the new
// decoder. The factory is also called for the initial
// decoder so the construction path is uniform.
val perSubscriptionDecoderFactory: () -> OpusDecoder = { decoderFactory(channelCount) }
val decoder = perSubscriptionDecoderFactory()
val player =
try {
playerFactory(channelCount)
@@ -1195,7 +1204,7 @@ class NestViewModel(
val isHushed = pubkey in _uiState.value.locallyHushed
val roomPlayer =
NestPlayer(
decoder = decoder,
initialDecoder = decoder,
player = player,
scope = viewModelScope,
// ~100 ms of audio buffered before the AudioTrack
@@ -1205,6 +1214,14 @@ class NestViewModel(
// tuned in the audio-rooms audit; see NestPlayer
// kdoc for details.
prerollFrames = ROOM_PLAYER_PREROLL_FRAMES,
// Trigger a decoder rebuild on every publisher
// boundary (re-issuing wrapper spliced in a new
// SUBSCRIBE → trackAlias changes). Without this,
// Opus's predictor state from the prior
// publisher's last frame is fed into the new
// publisher's first frame and produces audible
// warble at every cliff-recycle / hot-swap.
decoderFactory = perSubscriptionDecoderFactory,
)
// Apply current mute + per-speaker hush state before play()
// opens the device so the first frame respects them.
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt
+61 -1
View File
@@ -42,7 +42,7 @@ import kotlinx.coroutines.launch
* decoder. Idempotent.
*/
class NestPlayer(
private val decoder: OpusDecoder,
initialDecoder: OpusDecoder,
private val player: AudioPlayer,
private val scope: CoroutineScope,
/**
@@ -61,11 +61,44 @@ class NestPlayer(
* Default is `0` so existing tests stand without modification.
*/
private val prerollFrames: Int = 0,
/**
* Optional factory called on every detected publisher boundary
* (track-alias change in the inbound [MoqObject] stream) to mint a
* fresh [OpusDecoder]. Used by the listener wrapper's re-issuing
* subscription pump
* ([com.vitorpamplona.nestsclient.ReconnectingNestsListener.reissuingSubscribe]):
* each new SUBSCRIBE through the relay produces objects with a
* different `trackAlias`, but they're spliced into the same
* `SharedFlow` — without a decoder reset on the boundary, Opus's
* predictor state from the prior publisher's last frame is fed
* into the new publisher's first frame, producing an audible
* warble at every JWT-refresh hot-swap on the speaker side OR
* cliff-detector recycle on the listener side.
*
* Default `null` keeps the legacy behaviour (no boundary
* detection, decoder lives for the player's whole lifetime) so
* existing tests / callers that don't care about boundaries
* stand unchanged. Production callers in
* [com.vitorpamplona.amethyst.commons.viewmodels.NestViewModel.openSubscription]
* pass a closure that captures the per-subscription channel-count
* and rebuilds via `decoderFactory(channelCount)`.
*/
private val decoderFactory: (() -> OpusDecoder)? = null,
) {
init {
require(prerollFrames >= 0) { "prerollFrames must be >= 0, got $prerollFrames" }
}
/**
* Active decoder. Replaced on detected publisher boundary when
* [decoderFactory] is non-null. `var` so the boundary path can
* release + rebuild without changing the rest of the loop's
* decoder reference; the `private` confines mutation to this
* class, and the decode loop runs single-coroutine so no cross-
* thread visibility hazards.
*/
private var decoder: OpusDecoder = initialDecoder
private var job: Job? = null
private var stopped = false
@@ -157,6 +190,13 @@ class NestPlayer(
com.vitorpamplona.quartz.utils.Log
.d("NestPlay") { "NestPlayer beginPlayback returned" }
}
// Track-alias of the most recently observed object.
// A change signals a publisher boundary (re-issuing
// subscription wrapper spliced in a new SUBSCRIBE).
// Only consulted when [decoderFactory] is non-null;
// legacy callers without a factory keep the prior
// single-decoder behaviour.
var lastTrackAlias: Long? = null
try {
objects.collect { obj ->
receivedObjects += 1
@@ -165,6 +205,26 @@ class NestPlayer(
"NestPlayer received obj #$receivedObjects (decoded=$decodedFrames empty=$emptyDecodes enqueued=$enqueued playbackBegun=$playbackBegun)"
}
}
// Publisher-boundary detection: if the trackAlias
// changed since the last object AND we have a
// factory to mint a fresh decoder, release the
// current decoder + build a new one. Without
// this, Opus's predictor state from the prior
// publisher's last frame is fed into the new
// publisher's first frame, producing audible
// warble at every JWT-refresh hot-swap (speaker
// side) or cliff-detector recycle (listener side).
// The prior-trackAlias guard avoids a spurious
// rebuild on the very first frame.
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"
}
runCatching { decoder.release() }
decoder = factory()
}
lastTrackAlias = obj.trackAlias
val pcm =
try {
decoder.decode(obj.payload)
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt
+95 -5
View File
@@ -232,7 +232,7 @@ class NestPlayerTest {
val sut =
NestPlayer(
decoder = decoder,
initialDecoder = decoder,
player = player,
scope = this,
prerollFrames = 3,
@@ -289,7 +289,7 @@ class NestPlayerTest {
val sut =
NestPlayer(
decoder = decoder,
initialDecoder = decoder,
player = player,
scope = this,
prerollFrames = 5,
@@ -326,7 +326,7 @@ class NestPlayerTest {
val sut =
NestPlayer(
decoder = decoder,
initialDecoder = decoder,
player = player,
scope = this,
prerollFrames = 3,
@@ -340,17 +340,107 @@ class NestPlayerTest {
sut.stop()
}
@Test
fun publisher_boundary_rebuilds_decoder_when_factory_provided() =
runTest {
// 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 factory: () -> OpusDecoder = {
if (factoryCallCount.getAndIncrement() == 0) decoderA else decoderB
}
val player = FakeAudioPlayer()
val objects =
flowOf(
// First subscription cycle: trackAlias = 7
moqObject(byteArrayOf(0x01), trackAlias = 7L),
moqObject(byteArrayOf(0x02), trackAlias = 7L),
// Wrapper re-issued — new SUBSCRIBE produces a
// different trackAlias. Decoder MUST be rebuilt
// before the next decode runs.
moqObject(byteArrayOf(0x03), trackAlias = 8L),
moqObject(byteArrayOf(0x04), trackAlias = 8L),
)
val sut =
NestPlayer(
initialDecoder = factory(),
player = player,
scope = this,
decoderFactory = factory,
)
sut.play(objects)
testScheduler.advanceUntilIdle()
assertEquals(2, factoryCallCount.value, "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_no_op_when_factory_is_null() =
runTest {
// Without a factory, NestPlayer keeps the same decoder
// across trackAlias changes — backwards-compat path.
val decoder = FakeOpusDecoder { byteToShorts(it) }
val player = FakeAudioPlayer()
val objects =
flowOf(
moqObject(byteArrayOf(0x01), trackAlias = 7L),
moqObject(byteArrayOf(0x02), trackAlias = 8L),
)
val sut = NestPlayer(decoder, player, this)
sut.play(objects)
testScheduler.advanceUntilIdle()
assertEquals(0, decoder.releaseCount, "no boundary-driven release without a factory")
sut.stop()
assertEquals(1, decoder.releaseCount, "released exactly once on stop")
}
// -- helpers -----------------------------------------------------------
private fun moqObject(payload: ByteArray): MoqObject =
private fun moqObject(
payload: ByteArray,
trackAlias: Long = 1L,
): MoqObject =
MoqObject(
trackAlias = 1,
trackAlias = trackAlias,
groupId = 0,
objectId = 0,
publisherPriority = 0x80,
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(