From 4714e3c7236866cca6baa93369f494d6490c0bab Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 6 May 2026 18:00:52 +0000 Subject: [PATCH] fix(nests): T13 reset Opus decoder on publisher boundary in NestPlayer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../commons/viewmodels/NestViewModel.kt | 21 +++- .../nestsclient/audio/NestPlayer.kt | 62 ++++++++++- .../nestsclient/audio/NestPlayerTest.kt | 100 +++++++++++++++++- 3 files changed, 175 insertions(+), 8 deletions(-) diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt index cb6b85209..942765d62 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt @@ -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. diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt index d70137cbf..ffcbf883a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt @@ -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) diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt index 9d63e22ca..ddb4a75b4 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayerTest.kt @@ -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(