fix(nests): three more dropout sources from the post-audit review

1. Split AudioPlayer.start() into allocate + beginPlayback to restore the synchronous DeviceUnavailable error path that the previous pre-roll fix collapsed. - AudioPlayer gains `beginPlayback()` (default no-op) so test fakes / desktop sinks aren't forced to grow a method they don't need. - AudioTrackPlayer.start() now allocates the AudioTrack + audio- priority writer thread but does NOT call AudioTrack.play(); beginPlayback() flips the device into the playing state. AudioTrack in MODE_STREAM explicitly supports write() before play() per the platform docs, which is exactly the contract pre-roll wants. - NestPlayer.play() now calls player.start() synchronously (caller catches DeviceUnavailable + rolls back the slot like before) and defers beginPlayback() until pre-roll fills. 2. MediaCodecOpusDecoder: drain output + retry input dequeue before dropping a frame. The prior 10 ms input-buffer dequeue followed by an unconditional `return ShortArray(0)` turned every transient stall (thermal throttling, GC pause, output not yet pulled) into a 20 ms audio gap. Now we drain whatever output is ready — freeing input slots — then retry input dequeue with a 5 ms timeout. Only after both misses do we drop the packet, and even then we return any output samples that the drain produced so the player doesn't underrun on the back of one tight cycle. The decoder's drain logic is extracted into a `drainAvailableOutput` helper so both the pressure-relief path and the post-queue main drain share it. 3. ReconnectingNestsListener: exponential backoff for the opener- throws retry path. Replaces the flat 1 000 ms `SUBSCRIBE_RETRY_BACKOFF_MS` with 250 → 500 → 1 000 ms, reset on first successful subscribe. The 250 ms floor is well under moq-rs's typical announce- propagation latency (< 200 ms), so a subscribe-before-announce miss usually recovers fast enough that the wrapper SharedFlow's ~1.3 s buffer hides the gap entirely.
2026-05-05 12:17:59 +00:00
parent 076b301d84
commit e4e55d1df6
5 changed files with 192 additions and 56 deletions
@@ -64,6 +64,15 @@ import android.media.AudioFormat as AndroidAudioFormat
 * frame (~50 hops/sec/speaker) and contended with whatever else `Dispatchers.IO`
 * was running; an audio-priority dedicated thread gets reliable scheduling
 * and removes the contention.
+ *
+ * Two-phase startup: [start] allocates the AudioTrack (in stopped state)
+ * + spins up the writer thread. [beginPlayback] calls `AudioTrack.play()`
+ * to flip the device into the playing state. Splitting the two lets
+ * [NestPlayer] pre-roll several decoded frames into the AudioTrack's
+ * internal buffer BEFORE the hardware starts pulling samples, so playback
+ * begins with ~100 ms of buffered audio instead of underrunning on the
+ * first frame. AudioTrack in `MODE_STREAM` explicitly supports `write()`
+ * before `play()` per the platform docs — this is the intended pattern.
 */
 class AudioTrackPlayer(
    private val usage: Int = AudioAttributes.USAGE_MEDIA,
@@ -144,16 +153,11 @@ class AudioTrackPlayer(
                )
            }

-        try {
-            newTrack.play()
-        } catch (t: Throwable) {
-            runCatching { newTrack.release() }
-            throw AudioException(
-                AudioException.Kind.DeviceUnavailable,
-                "AudioTrack.play() rejected start",
-                t,
-            )
-        }
+        // NOTE: deliberately NOT calling `newTrack.play()` here — that's
+        // [beginPlayback]'s job. AudioTrack.write is legal in the not-yet-
+        // playing state for `MODE_STREAM`, which is the contract that lets
+        // [NestPlayer] pre-roll decoded frames into the buffer before the
+        // hardware starts consuming.
        applyMuteVolume(newTrack)
        // Spin up the audio-priority writer thread. The executor is private
        // to this player instance so per-speaker NestPlayer pumps don't
@@ -176,6 +180,29 @@ class AudioTrackPlayer(
        track = newTrack
    }

+    override fun beginPlayback() {
+        // Idempotent: AudioTrack.play() on an already-playing track is a
+        // no-op, and we gate on the track itself being non-null so a
+        // beginPlayback before start (or after stop) silently no-ops
+        // rather than blowing up — matches the rest of the player's
+        // tolerant-of-misordered-calls posture.
+        val t = track ?: return
+        try {
+            t.play()
+        } catch (e: Throwable) {
+            // PLAY-on-uninitialized AudioTrack is the only realistic
+            // failure here, and we already guard against an unallocated
+            // track above. Throw so [NestPlayer]'s outer catch surfaces
+            // it via `onError(AudioException.PlaybackFailed)` — same path
+            // that handles every other mid-stream device failure.
+            throw AudioException(
+                AudioException.Kind.DeviceUnavailable,
+                "AudioTrack.play() rejected start",
+                e,
+            )
+        }
+    }
+
    override suspend fun enqueue(pcm: ShortArray) {
        val t = track ?: throw AudioException(AudioException.Kind.PlaybackFailed, "player not started")
        val dispatcher =
@@ -57,8 +57,33 @@ class MediaCodecOpusDecoder : OpusDecoder {
    override fun decode(opusPacket: ByteArray): ShortArray {
        check(!released) { "decoder released" }

-        val inputIndex = codec.dequeueInputBuffer(DEQUEUE_TIMEOUT_US)
-        if (inputIndex < 0) return ShortArray(0)
+        // Pre-sized output. Write decoded samples directly into the array
+        // via ShortBuffer.get(dst, off, len) — the previous shape went
+        // through ArrayList<Short>, which boxed every PCM sample
+        // (~48 000 alloc/sec/speaker on the audio hot path).
+        val out = ShortArray(AudioFormat.FRAME_SIZE_SAMPLES)
+        var outPos = 0
+
+        // 1. Acquire an input slot. On rare back-pressure (output buffers
+        //    haven't been drained fast enough → all input slots are tied
+        //    up), drain whatever output IS ready first to free slots,
+        //    then retry input dequeue with a longer timeout. The previous
+        //    shape returned `ShortArray(0)` immediately when the first
+        //    10 ms dequeue missed, turning every transient stall (thermal
+        //    throttle, GC pause) into a 20 ms audio gap.
+        var inputIndex = codec.dequeueInputBuffer(DEQUEUE_TIMEOUT_US)
+        if (inputIndex < 0) {
+            outPos = drainAvailableOutput(out, outPos)
+            inputIndex = codec.dequeueInputBuffer(DEQUEUE_RETRY_TIMEOUT_US)
+            if (inputIndex < 0) {
+                // Genuine input starvation past the retry window. Drop
+                // this packet (presentationTimeUs is NOT advanced — Opus
+                // PLC will paper over the gap on the listener) and
+                // return whatever we drained from output so the player
+                // doesn't underrun on the back of one tight cycle.
+                return if (outPos > 0) out.copyOf(outPos) else ShortArray(0)
+            }
+        }
        val inputBuffer =
            codec.getInputBuffer(inputIndex)
                ?: throw AudioException(
@@ -71,17 +96,25 @@ class MediaCodecOpusDecoder : OpusDecoder {
        // Advance presentation time by one 20 ms frame.
        presentationTimeUs += FRAME_DURATION_US

-        // 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.
-        //
-        // 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
+        // 2. Drain whatever output is ready right now. A single Opus
+        //    packet typically yields exactly one output buffer, but
+        //    [drainAvailableOutput] tolerates an INFO_OUTPUT_FORMAT_CHANGED
+        //    that some devices emit before the first PCM frame.
+        outPos = drainAvailableOutput(out, outPos)
+        return if (outPos == out.size) out else out.copyOf(outPos)
+    }
+
+    /**
+     * Drain any output buffers MediaCodec has ready, copying samples into
+     * [out] starting at [startPos]. Returns the new write position.
+     * Stops on the first empty / EOS / TRY_AGAIN_LATER signal — never
+     * blocks past the [DEQUEUE_TIMEOUT_US] dequeue probe.
+     */
+    private fun drainAvailableOutput(
+        out: ShortArray,
+        startPos: Int,
+    ): Int {
+        var outPos = startPos
        var formatChangeAbsorbed = false
        drain@ while (true) {
            val outputIndex = codec.dequeueOutputBuffer(bufferInfo, DEQUEUE_TIMEOUT_US)
@@ -126,7 +159,7 @@ class MediaCodecOpusDecoder : OpusDecoder {
                }
            }
        }
-        return if (outPos == out.size) out else out.copyOf(outPos)
+        return outPos
    }

    override fun release() {
@@ -138,6 +171,20 @@ class MediaCodecOpusDecoder : OpusDecoder {

    companion object {
        private const val DEQUEUE_TIMEOUT_US = 10_000L // 10 ms
+
+        /**
+         * Fallback timeout after the first input dequeue misses + we
+         * drain pending output to free slots. Deliberately well under
+         * the 20 ms frame cadence so a tight back-pressure window is
+         * absorbed without falling off the per-frame budget. The
+         * previous code took a 0 ms retry (i.e. dropped the frame
+         * outright on the first miss), which turned every transient
+         * stall into ~20 ms of silence. 5 ms gives MediaCodec time to
+         * free a slot after the output drain without blowing the
+         * frame budget.
+         */
+        private const val DEQUEUE_RETRY_TIMEOUT_US = 5_000L // 5 ms
+
        private const val FRAME_DURATION_US = 20_000L // 20 ms

        private fun buildFormat(): MediaFormat {
@@ -340,6 +340,14 @@ private class ReconnectingHandle(
                        }
                    if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest

+                    // Backoff for the "opener threw" retry path — exponential
+                    // 250 → 500 → 1 000 ms with reset on success. The previous
+                    // shape used a flat 1 000 ms which, combined with the
+                    // 64-frame (~1.3 s) wrapper buffer, just barely hid the
+                    // first-retry gap; a quick retry usually succeeds because
+                    // moq-rs propagates announces in < 200 ms.
+                    var subscribeRetryDelayMs = SUBSCRIBE_RETRY_BACKOFF_INITIAL_MS
+
                    while (currentCoroutineContext().isActive) {
                        val handle =
                            try {
@@ -368,14 +376,21 @@ private class ReconnectingHandle(
                                // backoff used between publisher cycles.
                                Log.w("NestRx") {
                                    "ReconnectingHandle.opener threw ${t::class.simpleName}: ${t.message} " +
-                                        "— retrying after ${SUBSCRIBE_RETRY_BACKOFF_MS}ms"
+                                        "— retrying after ${subscribeRetryDelayMs}ms"
                                }
                                null
                            }
                        if (handle == null) {
-                            delay(SUBSCRIBE_RETRY_BACKOFF_MS)
+                            delay(subscribeRetryDelayMs)
+                            subscribeRetryDelayMs =
+                                (subscribeRetryDelayMs * 2)
+                                    .coerceAtMost(SUBSCRIBE_RETRY_BACKOFF_MAX_MS)
                            continue
                        }
+                        // Success: reset the backoff so the next publisher-
+                        // cycle gap starts at the floor again (rather than
+                        // inheriting the last failure window's saturation).
+                        subscribeRetryDelayMs = SUBSCRIBE_RETRY_BACKOFF_INITIAL_MS
                        liveHandleRef.set(handle)
                        try {
                            handle.objects.collect { frames.emit(it) }
@@ -427,14 +442,24 @@ private class ReconnectingHandle(
        // gone publisher doesn't spin the relay with re-subscribes.
        private const val RESUBSCRIBE_BACKOFF_MS = 100L

-        // Backoff after an opener throws (typically: subscribe-before-
-        // announce arrives at the relay before the publisher exists,
-        // and the relay FINs the bidi without a SubscribeOk/Drop). One
-        // second is well over moq-rs's typical announce-propagation
-        // latency (< 200 ms in production traces), so the next retry
-        // usually succeeds; long enough that a never-arriving publisher
-        // doesn't hammer the relay either.
-        private const val SUBSCRIBE_RETRY_BACKOFF_MS = 1_000L
+        // Exponential backoff for the opener-throws retry path.
+        // Typical case: subscribe-before-announce arrives at the relay
+        // before the publisher exists, and the relay FINs the bidi
+        // without a SubscribeOk/Drop reply.
+        //
+        // - INITIAL = 250 ms: under moq-rs's typical announce-propagation
+        //   latency (< 200 ms in production traces), so the very first
+        //   retry usually succeeds and the listener-side buffer hides
+        //   the gap.
+        // - Doubles each miss → 500 ms → 1 000 ms.
+        // - MAX = 1 000 ms: matches the previous flat constant; long
+        //   enough that a never-arriving publisher doesn't hammer the
+        //   relay, short enough to stay under the wrapper SharedFlow's
+        //   ~1.3 s buffer once playback is rolling on the next attempt.
+        // - Reset on first successful subscribe so a subsequent
+        //   publisher-cycle gap doesn't inherit the previous saturation.
+        private const val SUBSCRIBE_RETRY_BACKOFF_INITIAL_MS = 250L
+        private const val SUBSCRIBE_RETRY_BACKOFF_MAX_MS = 1_000L

        private val SYNTH_OK =
            SubscribeOk(
@@ -99,11 +99,37 @@ interface AudioCapture {
 /**
 * Sink for PCM audio playback. Implementations buffer internally — [enqueue]
 * may suspend if the device's playback buffer is full.
+ *
+ * **Two-phase startup.** [start] allocates the underlying device + per-instance
+ * resources but does NOT begin consuming samples; [beginPlayback] flips the
+ * device into the playing state. Splitting the two lets [com.vitorpamplona.nestsclient.audio.NestPlayer]
+ * pre-roll a few decoded frames into the device's buffer before playback
+ * starts, so the AudioTrack has slack the moment the hardware begins pulling
+ * samples. Calling [enqueue] between [start] and [beginPlayback] is allowed
+ * and is the intended pattern. Implementations that don't need the
+ * distinction (test fakes, software-only sinks) can leave the default
+ * [beginPlayback] no-op alone.
 */
 interface AudioPlayer {
-    /** Allocate underlying audio resources and begin playback. */
+    /**
+     * Allocate underlying audio resources. The returned device is in a
+     * "ready, not playing" state — [enqueue] is allowed but the hardware
+     * doesn't consume samples until [beginPlayback] is called. Throws on
+     * device-unavailable so callers (typically
+     * [com.vitorpamplona.nestsclient.audio.NestPlayer.play]) get a
+     * synchronous failure they can roll back the subscription on.
+     */
    fun start()

+    /**
+     * Transition the allocated device into the playing state. The
+     * hardware begins pulling samples from whatever's already been
+     * [enqueue]d. Default no-op so test fakes / software-only sinks
+     * don't have to grow a method they'll never use; production
+     * Android implementations override to call `AudioTrack.play()`.
+     */
+    fun beginPlayback() {}
+
    /**
     * Feed one PCM frame (any length, but typically [AudioFormat.FRAME_SIZE_SAMPLES]
     * samples) into the playback queue.
@@ -93,30 +93,41 @@ class NestPlayer(
        check(!stopped) { "NestPlayer already stopped" }
        check(job == null) { "NestPlayer.play already called" }

-        // Pre-roll buffer holds decoded PCM until either [prerollFrames]
-        // frames have arrived or the upstream flow ends. The underlying
-        // [AudioPlayer] is only started once we have something to flush,
-        // so a flow that never produces PCM (decoder always empty, no
-        // audio in the room) never opens the device.
+        // Allocate the device synchronously so a [AudioException.DeviceUnavailable]
+        // (audio policy denial, AudioTrack rejected, etc.) propagates to
+        // the caller — typically `NestViewModel.openSubscription`, which
+        // catches and rolls back the freshly-reserved subscription slot.
+        // Routing this failure through `onError` instead would attach the
+        // slot first and leave a permanent "Connecting…" spinner on the
+        // speaker tile when the device fails to allocate.
        //
-        // Note: `player.start()` is now deferred into the launch body. If
-        // it throws, the failure is reported via `onError` rather than
-        // propagating synchronously to `play()`'s caller. This is
-        // intentional — the device-allocation cost was previously paid
-        // up-front and amplified perceived join latency; pushing it
-        // behind the first decoded frame both lets pre-roll work and
-        // matches the rest of the pipeline's "audible-failure-via-onError"
-        // contract.
+        // Two-phase startup: [AudioPlayer.start] allocates without
+        // beginning playback; [AudioPlayer.beginPlayback] flips the
+        // device into the playing state. We delay [beginPlayback] until
+        // the pre-roll buffer is full so the first frames already
+        // populate the device's internal buffer when the hardware
+        // starts pulling samples.
+        player.start()
+
+        // Pre-roll buffer holds decoded PCM until either [prerollFrames]
+        // frames have arrived or the upstream flow ends. Once the
+        // threshold is met (or the flow ends with anything queued), we
+        // call [AudioPlayer.beginPlayback] and flush the buffer in a
+        // tight loop so the device starts playback with a populated
+        // buffer. A flow that never produces PCM (decoder always empty,
+        // no audio in the room) never calls [beginPlayback] — the
+        // allocated device sits in the "ready, not playing" state until
+        // [stop] tears it down.
        job =
            scope.launch {
                val preroll = ArrayDeque<ShortArray>(prerollFrames.coerceAtLeast(1))
-                var started = false
+                var playbackBegun = false

-                suspend fun startAndFlushIfNeeded() {
-                    if (started) return
+                suspend fun beginAndFlushIfNeeded() {
+                    if (playbackBegun) return
                    if (preroll.isEmpty()) return
-                    player.start()
-                    started = true
+                    player.beginPlayback()
+                    playbackBegun = true
                    while (preroll.isNotEmpty()) {
                        player.enqueue(preroll.removeFirst())
                    }
@@ -140,12 +151,12 @@ class NestPlayer(
                            }
                        if (pcm.isNotEmpty()) {
                            onLevel(peakAmplitude(pcm))
-                            if (started) {
+                            if (playbackBegun) {
                                player.enqueue(pcm)
                            } else {
                                preroll.addLast(pcm)
                                if (preroll.size >= prerollFrames) {
-                                    startAndFlushIfNeeded()
+                                    beginAndFlushIfNeeded()
                                }
                            }
                        }
@@ -154,7 +165,7 @@ class NestPlayer(
                    // (e.g. the publisher cycled before pre-roll was full,
                    // or the room ended). Flush whatever's queued so any
                    // already-decoded audio still reaches the device.
-                    startAndFlushIfNeeded()
+                    beginAndFlushIfNeeded()
                } catch (ce: CancellationException) {
                    throw ce
                } catch (t: Throwable) {