From 076b301d84a1fb64fa5ff0b1edddadd7a8c28c0e Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 5 May 2026 03:06:40 +0000 Subject: [PATCH] fix(nests): close 4 audio-dropout sources across listener + speaker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. Listener-side pre-roll + bigger playback buffer + audio-priority thread. - NestPlayer buffers 5 decoded frames (~100 ms) before starting the AudioPlayer, masking the first-frame underrun that fires whenever Compose / GC briefly stalls Main. - AudioTrackPlayer sizes the AudioTrack at max(minBuffer*16, 250 ms) instead of minBuffer*4 (~80 ms) and writes via a per-instance audio-priority single-thread executor (Process.THREAD_PRIORITY_AUDIO) instead of Dispatchers.IO, so WRITE_BLOCKING never contends with unrelated IO work. 2. MoqLiteSession.subscribe: hoist response typeCode out of the collect lambda. readVarint advances pos permanently while readSizePrefixed only rolls back its own length-varint, so a chunk-split between type and body would re-read the body's size prefix as the type code on the next chunk and misframe the response. Mirrors the same fix already in handleInboundBidi. 3. MoqLiteSession.subscribe: register the subscription in the map BEFORE writing the SUBSCRIBE bytes on the wire. The relay can open the first group's uni stream before our continuation re-enters [state] to register; if so, drainOneGroup looked the id up against an empty map and silently dropped the frame (~1 frame / 20 ms gap on first attach). Wrap the post-register writes so a transport-failure unwinds the orphan registration. 4. Hot-swap moq-lite publisher across JWT-refresh boundaries. - NestMoqLiteBroadcaster: publisher is now @Volatile + supports swapPublisher(); capture loop snapshots the reference per frame and resets framesInCurrentGroup on swap. - MoqLiteNestsSpeaker implements a new internal HotSwappablePublisherSource interface that exposes openPublisherForHotSwap(track) without spinning up a broadcaster. - ReissuingBroadcastHandle keeps a single long-lived broadcaster across session recycles when the speaker supports hot swap; legacy IETF / fake speakers fall back to close-then-restart. - connectReconnectingNestsSpeaker.orchestrator hoists the old- session close onto a sibling launch so the wrapper can swap the publisher into the broadcaster on the new session before the old session's WebTransport drops. Eliminates the previously- accepted 50–150 ms audible silence at every JWT refresh. --- .../commons/viewmodels/NestViewModel.kt | 25 +- .../nestsclient/audio/AudioTrackPlayer.kt | 84 ++++- .../nestsclient/MoqLiteNestsSpeaker.kt | 60 ++- .../nestsclient/ReconnectingNestsSpeaker.kt | 354 +++++++++++++----- .../audio/NestMoqLiteBroadcaster.kt | 91 ++++- .../nestsclient/audio/NestPlayer.kt | 61 ++- .../nestsclient/moq/lite/MoqLiteSession.kt | 67 +++- 7 files changed, 618 insertions(+), 124 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 896c7398b..183bc9d27 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 @@ -868,7 +868,19 @@ class NestViewModel( try { val isMuted = _uiState.value.isMuted val isHushed = pubkey in _uiState.value.locallyHushed - val roomPlayer = NestPlayer(decoder, player, viewModelScope) + val roomPlayer = + NestPlayer( + decoder = decoder, + player = player, + scope = viewModelScope, + // ~100 ms of audio buffered before the AudioTrack + // starts consuming. Long enough to hide a typical + // Main-thread hiccup (Compose recomposition / GC) + // without making the join feel laggy. Empirically + // tuned in the audio-rooms audit; see NestPlayer + // kdoc for details. + prerollFrames = ROOM_PLAYER_PREROLL_FRAMES, + ) // Apply current mute + per-speaker hush state before play() // opens the device so the first frame respects them. player.setMutedSafe(isMuted) @@ -1246,6 +1258,17 @@ sealed class BroadcastUiState { */ const val SPEAKING_TIMEOUT_MS: Long = 250L +/** + * Per-speaker pre-roll: number of decoded PCM frames buffered before the + * underlying [com.vitorpamplona.nestsclient.audio.AudioPlayer] starts + * consuming. 5 × 20 ms ≈ 100 ms of audio — long enough to mask a typical + * Main-thread stall (Compose recomposition / GC) without adding perceptible + * join latency. Combines with [com.vitorpamplona.nestsclient.audio.AudioTrackPlayer]'s + * ~250 ms AudioTrack buffer for ~350 ms of total slack between the + * arrival-of-frame and the underrun horizon. + */ +const val ROOM_PLAYER_PREROLL_FRAMES: Int = 5 + /** * Coalescing interval for [NestViewModel.audioLevels]. The decode loop * pushes a fresh peak every ~20 ms (one per Opus frame); we publish to diff --git a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt index 01dfddd07..e9c0e62f1 100644 --- a/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt +++ b/nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt @@ -22,8 +22,12 @@ package com.vitorpamplona.nestsclient.audio import android.media.AudioAttributes import android.media.AudioTrack -import kotlinx.coroutines.Dispatchers +import android.os.Process +import kotlinx.coroutines.ExecutorCoroutineDispatcher +import kotlinx.coroutines.asCoroutineDispatcher import kotlinx.coroutines.withContext +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors import android.media.AudioFormat as AndroidAudioFormat /** @@ -45,9 +49,21 @@ import android.media.AudioFormat as AndroidAudioFormat * `MediaRecorder.AudioSource.VOICE_COMMUNICATION` regardless of the * playback usage. * - * Buffer sizing: 4× minimum so the producer can fall behind by ~80 ms before - * dropouts, which roughly matches the jitter the WebTransport datagram path - * introduces over typical mobile networks. + * Buffer sizing: target ~250 ms of slack, computed as + * `max(minBuffer * 16, 250 ms-equivalent)`. The previous 4× minimum (~80 ms + * by the device-reported floor) underran on devices with very small + * `getMinBufferSize` returns and on any handset whose decode loop got + * stalled by Compose recomposition / GC on Main. 250 ms matches the + * jitter-buffer depth Spaces / Clubhouse use for hands-free audio rooms, + * and combined with the per-subscription pre-roll in [NestPlayer] keeps + * the AudioTrack from underruning across typical mobile network jitter. + * + * Threading: writes go through a per-instance audio-priority single-thread + * dispatcher (`HandlerThread` + [Process.THREAD_PRIORITY_AUDIO]) instead of + * `Dispatchers.IO`. The previous shape did one IO dispatcher hop per Opus + * 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. */ class AudioTrackPlayer( private val usage: Int = AudioAttributes.USAGE_MEDIA, @@ -57,6 +73,18 @@ class AudioTrackPlayer( private var muted: Boolean = false private var volume: Float = 1f + /** + * Dedicated audio-priority single-thread executor for AudioTrack writes. + * Lazily created on [start] and shut down on [stop] so a never-started + * player doesn't leak a thread. `THREAD_PRIORITY_AUDIO` is the standard + * Linux nice level for VoIP / WebRTC playback paths on Android — it sits + * above background but below true audio-callback priority, so the OS + * scheduler keeps it running through GC / Compose recomposition without + * starving other threads. + */ + private var audioExecutor: ExecutorService? = null + private var audioDispatcher: ExecutorCoroutineDispatcher? = null + override fun start() { if (track != null) return @@ -79,7 +107,14 @@ class AudioTrackPlayer( "AudioTrack.getMinBufferSize returned $minBuffer for ${AudioFormat.SAMPLE_RATE_HZ} Hz", ) } - val bufferBytes = minBuffer * 4 + // Target ~250 ms of audio: enough headroom so the decode loop can + // miss its 20 ms cadence by an order of magnitude before the device + // underruns. Take the larger of `minBuffer * 16` and an explicit + // 250 ms-equivalent so devices that report a small minBuffer still + // get the same wall-clock slack. + val targetBytes250Ms = + (AudioFormat.SAMPLE_RATE_HZ / 4) * AudioFormat.BYTES_PER_SAMPLE * AudioFormat.CHANNELS + val bufferBytes = maxOf(minBuffer * 16, targetBytes250Ms) val newTrack = try { @@ -120,14 +155,37 @@ class AudioTrackPlayer( ) } applyMuteVolume(newTrack) + // Spin up the audio-priority writer thread. The executor is private + // to this player instance so per-speaker NestPlayer pumps don't + // contend on a shared queue. Priority is set inside the thread's + // Runnable because Linux thread priority is per-OS-thread, not + // per-Java Thread; `Thread.setPriority` does NOT translate to a + // Linux nice level on Android. `Process.setThreadPriority` does. + val executor = + Executors.newSingleThreadExecutor { r -> + Thread( + { + runCatching { Process.setThreadPriority(Process.THREAD_PRIORITY_AUDIO) } + r.run() + }, + "nest-audio-writer", + ) + } + audioExecutor = executor + audioDispatcher = executor.asCoroutineDispatcher() track = newTrack } override suspend fun enqueue(pcm: ShortArray) { val t = track ?: throw AudioException(AudioException.Kind.PlaybackFailed, "player not started") - // AudioTrack.write blocks if the internal buffer is full. Run on IO so - // we don't stall a coroutine dispatcher backed by a small thread pool. - withContext(Dispatchers.IO) { + val dispatcher = + audioDispatcher + ?: throw AudioException(AudioException.Kind.PlaybackFailed, "audio dispatcher not initialized") + // AudioTrack.write blocks if the internal buffer is full. Run on the + // per-instance audio-priority writer thread so the WRITE_BLOCKING + // suspension is on a thread the OS schedules tightly, and so we + // don't compete with whatever else `Dispatchers.IO` is running. + withContext(dispatcher) { val written = t.write(pcm, 0, pcm.size, AudioTrack.WRITE_BLOCKING) if (written < 0) { throw AudioException( @@ -155,6 +213,16 @@ class AudioTrackPlayer( runCatching { t.flush() } runCatching { t.stop() } runCatching { t.release() } + // Tear down the audio-priority writer. close() on the + // ExecutorCoroutineDispatcher shuts down the executor (and the + // underlying single thread) — pending writes are abandoned rather + // than blocked on, since the AudioTrack itself has already been + // stopped + released two lines up so any further `write` would + // throw IllegalStateException anyway. + audioDispatcher?.close() + audioDispatcher = null + audioExecutor?.let { runCatching { it.shutdownNow() } } + audioExecutor = null } private fun applyMuteVolume(track: AudioTrack) { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt index b1e9b3dac..c719a8cac 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt @@ -59,7 +59,8 @@ class MoqLiteNestsSpeaker internal constructor( * Defaults to [NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP]. */ private val framesPerGroup: Int = NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP, -) : NestsSpeaker { +) : NestsSpeaker, + HotSwappablePublisherSource { override val state: StateFlow = mutableState.asStateFlow() private val gate = Mutex() @@ -100,7 +101,7 @@ class MoqLiteNestsSpeaker internal constructor( NestMoqLiteBroadcaster( capture = captureFactory(), encoder = encoderFactory(), - publisher = publisher, + initialPublisher = publisher, scope = scope, framesPerGroup = framesPerGroup, ).also { @@ -113,7 +114,7 @@ class MoqLiteNestsSpeaker internal constructor( // the session — without this signal the // outward state stays on Broadcasting // and the room is silently mute. - onBroadcastTerminalFailure() + reportBroadcastTerminalFailure() }, onLevel = onLevel, ) @@ -139,6 +140,15 @@ class MoqLiteNestsSpeaker internal constructor( } } + /** + * [HotSwappablePublisherSource] implementation. See the interface + * kdoc — this method mints a fresh publisher on the session + * WITHOUT spinning up a broadcaster on top of it. Used by the + * reconnect wrapper's hot-swap path; not called from the + * non-reconnecting path which goes through [startBroadcasting]. + */ + override suspend fun openPublisherForHotSwap(track: String): MoqLitePublisherHandle = session.publish(broadcastSuffix = speakerPubkeyHex, track = track) + /** * Compare-and-clear that runs from inside [close] (already holds * [gate]) and from [MoqLiteBroadcastHandle.close] (doesn't). @@ -160,8 +170,13 @@ class MoqLiteNestsSpeaker internal constructor( * the reconnect orchestrator (`ReconnectingNestsSpeaker`) observes * a terminal state and recycles the session. No-op if the speaker * is already in a terminal state. + * + * Also exposed via [HotSwappablePublisherSource.reportBroadcastTerminalFailure] + * so the hot-swap pump (which owns its own long-lived broadcaster) + * can drive the same orchestrator-reconnect path the legacy + * `startBroadcasting` flow does. */ - private fun onBroadcastTerminalFailure() { + override fun reportBroadcastTerminalFailure() { val current = mutableState.value if (current is NestsSpeakerState.Failed || current is NestsSpeakerState.Closed) return mutableState.value = @@ -257,3 +272,40 @@ internal class MoqLiteBroadcastHandle( parent.broadcastClosed(this) } } + +/** + * Internal hot-swap seam: speakers that expose this interface let the + * reconnect wrapper retarget a long-lived + * [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster] onto a + * freshly-opened moq-lite session's publisher without restarting the + * AudioRecord / Opus encoder pipeline. Implemented by + * [MoqLiteNestsSpeaker]; not implemented by the IETF reference + * [DefaultNestsSpeaker], which falls back to the close-then-restart path + * inside [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker]. + * + * The wrapper uses an `as?` cast to detect support so this interface + * can stay package-internal — protocol consumers never see it. + */ +internal interface HotSwappablePublisherSource { + /** + * Open a fresh [MoqLitePublisherHandle] on the underlying moq-lite + * session. Caller owns the returned handle's lifetime (typically + * via [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster.swapPublisher]'s + * close-the-old contract). + */ + suspend fun openPublisherForHotSwap(track: String): MoqLitePublisherHandle + + /** + * Surface a broadcast-pipeline terminal failure (e.g. sustained + * `publisher.send` errors past + * [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS]) + * by flipping the speaker's state to [NestsSpeakerState.Failed]. + * Called by the hot-swap pump when the long-lived broadcaster's + * `onTerminalFailure` fires; lets the reconnect orchestrator + * observe the terminal state and recycle the session, matching + * the legacy + * [MoqLiteNestsSpeaker.startBroadcasting] path's failure + * propagation. + */ + fun reportBroadcastTerminalFailure() +} diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt index a7f627ef0..e22039cd1 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt @@ -21,6 +21,7 @@ package com.vitorpamplona.nestsclient import com.vitorpamplona.nestsclient.audio.AudioCapture +import com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster import com.vitorpamplona.nestsclient.audio.OpusEncoder import com.vitorpamplona.nestsclient.transport.WebTransportFactory import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner @@ -68,15 +69,23 @@ import java.util.concurrent.atomic.AtomicReference * `close` cancels the re-issue pump and best-effort closes the * latest live handle. * - * **Audio gap during refresh** — the wrapper closes the current - * underlying speaker (which stops the mic capture + Opus encoder + - * publisher) before opening the next, so the listener side will - * hear ~50–150 ms of silence at each recycle boundary. That's the - * trade-off we pay for a clean session swap; the alternative - * (carrying the mic capture across sessions) would require deeper - * plumbing into the audio pipeline. Acceptable for v1 — - * 9-min-spaced 150 ms gaps are well below the noise floor of a - * voice call. + * **Audio gap during refresh — eliminated for moq-lite** — when the + * underlying speaker implements [HotSwappablePublisherSource] + * (which [MoqLiteNestsSpeaker] does), the wrapper keeps a single + * long-lived [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster] + * alive across session recycles and only swaps the + * [com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle] + * underneath it. The mic + encoder run continuously through the + * boundary and the listener hears no silence. The old session + * close runs in parallel with the new openOnce so the WebTransport + * teardown doesn't block the swap. + * + * **Audio gap during refresh — legacy path** — for speakers that + * don't implement [HotSwappablePublisherSource] (the IETF reference + * `DefaultNestsSpeaker` and test fakes), the wrapper falls back to + * close-then-restart: the listener hears ~50–150 ms of silence per + * recycle. Acceptable because the IETF path is reference-only — + * production runs the moq-lite hot-swap path. * * Cancellation: cancelling [scope] (typically the room screen's VM * scope) cancels the reconnect loop and closes both the active @@ -186,16 +195,41 @@ suspend fun connectReconnectingNestsSpeaker( } if (terminal == null) { // Refresh deadline hit before any terminal state — - // planned recycle, not a failure. Close the old - // speaker; don't bump `attempt` (it's not a - // backoff event) so the next openOnce() runs - // immediately. - try { - speaker.close() - } catch (ce: kotlinx.coroutines.CancellationException) { - throw ce - } catch (_: Throwable) { - // Best-effort. + // planned recycle, not a failure. Don't bump + // `attempt` (it's not a backoff event) so the next + // openOnce() runs immediately. + // + // Hand the old speaker's close to a separate + // launch so it runs IN PARALLEL with the next + // iteration's openOnce(). Sequence on the wire: + // + // 1. (concurrent) old speaker close — closes + // old session's publisher + WebTransport. + // 2. (concurrent) new openOnce() → sets + // activeSpeaker = newSpeaker. + // 3. Wrapper's hot-swap pump observes the + // activeSpeaker change, opens a publisher + // on the new session, swaps it into the + // long-lived broadcaster, closes the old + // publisher. + // + // The previous shape closed the old speaker + // synchronously BEFORE openOnce, which forced the + // mic + encoder to stop and re-open across the + // boundary (50–150 ms audible silence at the + // listener). With the close hoisted onto a + // sibling job, the broadcaster keeps capturing + // through the recycle and the listener hears + // continuous audio. + val toClose = speaker + scope.launch { + try { + toClose.close() + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + // Best-effort. + } } attempt = 0 refreshTriggered = true @@ -249,7 +283,14 @@ suspend fun connectReconnectingNestsSpeaker( throw NestsException(firstReady.reason, firstReady.cause) } - return ReconnectingSpeakerHandle(state, activeSpeaker, orchestrator, scope) + return ReconnectingSpeakerHandle( + mutableState = state, + activeSpeaker = activeSpeaker, + orchestrator = orchestrator, + scope = scope, + captureFactory = captureFactory, + encoderFactory = encoderFactory, + ) } private fun isUserCancelledSpeaker(state: NestsSpeakerState.Failed): Boolean { @@ -266,6 +307,17 @@ private class ReconnectingSpeakerHandle( private val activeSpeaker: MutableStateFlow, private val orchestrator: Job, private val scope: CoroutineScope, + /** + * Audio-pipeline factories propagated from + * [connectReconnectingNestsSpeaker]. Used by the hot-swap + * [ReissuingBroadcastHandle] to construct ONE long-lived + * [NestMoqLiteBroadcaster] that survives session recycles — + * without these the broadcaster would have to be re-built + * (which restarts the mic + encoder + adds a 50–150 ms + * audible gap) on every JWT refresh. + */ + private val captureFactory: () -> AudioCapture, + private val encoderFactory: () -> OpusEncoder, ) : NestsSpeaker { override val state: StateFlow = mutableState.asStateFlow() @@ -291,9 +343,16 @@ private class ReconnectingSpeakerHandle( ?: error("no live session — wait for state == Connected before startBroadcasting") val handle = - ReissuingBroadcastHandle(activeSpeaker, scope, onLevel) { closed -> - if (activeBroadcast === closed) activeBroadcast = null - } + ReissuingBroadcastHandle( + activeSpeaker = activeSpeaker, + scope = scope, + captureFactory = captureFactory, + encoderFactory = encoderFactory, + onLevel = onLevel, + onClose = { closed -> + if (activeBroadcast === closed) activeBroadcast = null + }, + ) handle.start() activeBroadcast = handle handle @@ -310,26 +369,42 @@ private class ReconnectingSpeakerHandle( } /** - * Stable [BroadcastHandle] backed by a re-issuing pump. Each time - * the wrapper opens a fresh session the pump cancels its prior - * iteration, calls [NestsSpeaker.startBroadcasting] on the new - * session, replays the cached mute intent on the resulting - * underlying handle, and parks until the next session swap. + * Stable [BroadcastHandle] backed by a re-issuing pump. Two paths, + * picked per-iteration based on whether the active speaker exposes + * the [HotSwappablePublisherSource] hook: * - * `setMuted` updates the cached intent unconditionally and forwards - * to whichever live underlying handle exists at the time. If no - * underlying handle is up (e.g. a brief gap during recycle), the - * intent is replayed on the next handle the pump opens, so the - * user-observed mute state is monotonic across recycles. + * - **Hot-swap path** (moq-lite): the pump constructs ONE + * long-lived [NestMoqLiteBroadcaster] on the first session + * and, on every subsequent session swap, opens a fresh + * [com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle] + * on the new session and atomically swaps it into the + * broadcaster via + * [NestMoqLiteBroadcaster.swapPublisher]. The mic + + * encoder + capture loop keep running through the swap, so + * listeners hear ZERO gap at the JWT-refresh boundary + * (vs the legacy ~50–150 ms close-then-restart silence). + * + * - **Legacy path** (IETF reference / test fakes): each session + * swap closes the prior `BroadcastHandle` from + * [NestsSpeaker.startBroadcasting] and opens a fresh one. Same + * behaviour as before this hot-swap was introduced. + * + * `setMuted` updates the cached intent unconditionally; the pump + * applies the latest intent to whichever underlying broadcaster / + * handle is currently live, so user-observed mute is monotonic + * across recycles. */ private class ReissuingBroadcastHandle( private val activeSpeaker: StateFlow, private val scope: CoroutineScope, + private val captureFactory: () -> AudioCapture, + private val encoderFactory: () -> OpusEncoder, /** - * Forwarded to every underlying [NestsSpeaker.startBroadcasting] - * the pump opens, so the local-speaking ring keeps animating - * across session recycles. Mirrors how [desiredMuted] is - * replayed — the user-observed signal is monotonic. + * Forwarded to the underlying broadcaster (hot-swap path) or + * `sp.startBroadcasting` (legacy path) so the local-speaking ring + * keeps animating across session recycles. Mirrors how + * [desiredMuted] is replayed — the user-observed signal is + * monotonic. */ private val onLevel: (Float) -> Unit, private val onClose: (ReissuingBroadcastHandle) -> Unit, @@ -337,17 +412,22 @@ private class ReissuingBroadcastHandle( @Volatile private var desiredMuted: Boolean = false @Volatile private var closed: Boolean = false + + /** Hot-swap path's long-lived broadcaster. Null until the first session arrives. */ + @Volatile private var hotSwapBroadcaster: NestMoqLiteBroadcaster? = null + + /** Legacy path's per-session handle. Cleared when the session swaps. */ private val liveHandle = AtomicReference(null) private var pumpJob: Job? = null override val isMuted: Boolean get() = desiredMuted fun start() { - // Re-broadcast pump: every time activeSpeaker changes, drop - // the prior broadcast (collectLatest cancels the inner - // body via awaitCancellation) and open a new one against - // the fresh session. The pattern mirrors the listener's - // SubscribeHandle re-issuance pump. + // Per-iteration pump: every time activeSpeaker changes, decide + // whether the new speaker supports hot swap. If yes, retarget + // the long-lived broadcaster onto its publisher; if no, fall + // back to the close-then-restart path the previous version + // used uniformly. pumpJob = scope.launch { activeSpeaker.collectLatest { sp -> @@ -367,60 +447,151 @@ private class ReissuingBroadcastHandle( return@collectLatest } if (closed) return@collectLatest - val handle = - try { - sp.startBroadcasting(onLevel) - } catch (ce: kotlinx.coroutines.CancellationException) { - // Don't `return@collectLatest` on cancel — - // propagate so the launched pumpJob actually - // dies on close/scope cancellation. The old - // `runCatching` shape ate the cancel. - throw ce - } catch (_: Throwable) { - null - } ?: return@collectLatest - if (closed) { - runCatching { handle.close() } - return@collectLatest - } - // Apply current mute intent BEFORE storing the - // handle so a setMuted that races us applies - // exactly once: either (a) we set intent → - // apply intent → store, and the racing setMuted - // sees the live handle and applies again (no-op - // on the broadcaster); or (b) the racing - // setMuted updates intent → we read intent → - // apply. Order doesn't matter; idempotent. - if (desiredMuted) { - runCatching { handle.setMuted(true) } - } - liveHandle.set(handle) - try { - // Park until activeSpeaker emits a new value - // (collectLatest cancels us) or close() runs - // (pumpJob.cancel). - awaitCancellation() - } finally { - // Clear our slot only if we still own it — - // close() may have already swapped in null. - if (liveHandle.get() === handle) liveHandle.set(null) - // Best-effort close on the way out: the user - // may have called wrapper.close (closed=true, - // pump cancelling), or activeSpeaker swapped - // (the prior speaker is about to be closed - // by the orchestrator anyway, but defensively - // closing here releases the broadcaster + - // publisher promptly rather than waiting for - // the speaker.close()). - runCatching { handle.close() } + + val hotSwap = sp as? HotSwappablePublisherSource + if (hotSwap != null) { + runHotSwapIteration(hotSwap) + } else { + runLegacyIteration(sp) } } } } + /** + * Hot-swap iteration body. On the first session: open a publisher, + * construct the long-lived broadcaster, start it. On subsequent + * sessions: open a publisher on the new session, swap into the + * existing broadcaster, close the OLD publisher (FINs its + * announce + group streams on the about-to-die session). + * + * Parks via [awaitCancellation] so [collectLatest] can cancel us + * cleanly when the next session arrives. The broadcaster is NOT + * stopped here — it lives across iterations and is only stopped + * when the wrapper itself closes. + */ + private suspend fun runHotSwapIteration(hotSwap: HotSwappablePublisherSource) { + val newPublisher = + try { + hotSwap.openPublisherForHotSwap(MoqLiteNestsListener.AUDIO_TRACK) + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + // Couldn't mint a publisher on this session (rare — + // the session is already past Connected). Bail; the + // next session swap will retry. + return + } + if (closed) { + runCatching { newPublisher.close() } + return + } + + val existing = hotSwapBroadcaster + if (existing == null) { + // First-ever iteration: build the broadcaster and start it. + val broadcaster = + NestMoqLiteBroadcaster( + capture = captureFactory(), + encoder = encoderFactory(), + initialPublisher = newPublisher, + scope = scope, + framesPerGroup = NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP, + ) + try { + broadcaster.start( + onTerminalFailure = { + // Bubble up via the speaker's state machine so + // the orchestrator's normal Failed-handling + // path recycles the session. Same shape the + // legacy `startBroadcasting`-internal path uses. + runCatching { hotSwap.reportBroadcastTerminalFailure() } + }, + onLevel = onLevel, + ) + } catch (t: Throwable) { + runCatching { newPublisher.close() } + throw t + } + // Apply current mute intent (a setMuted that arrived before + // the broadcaster existed has already updated desiredMuted). + if (desiredMuted) broadcaster.setMuted(true) + hotSwapBroadcaster = broadcaster + } else { + // Subsequent iteration: hot swap. The capture / encoder + // pipeline is already running and feeding [existing.publisher] + // (the soon-to-be-old one). Install the new publisher, + // grab the old, close it. The capture loop's next snapshot + // picks up the new publisher and resets its group counter. + val old = existing.swapPublisher(newPublisher) + // Re-apply mute on the broadcaster — it survives swap, but + // a setMuted that arrived during the gap between the last + // session's terminal state and this swap may have flipped + // [desiredMuted] without finding a live broadcaster (no, it + // would have seen [hotSwapBroadcaster] since that's + // long-lived; but the no-op on equality keeps this safe). + existing.setMuted(desiredMuted) + // Close the old publisher AFTER the broadcaster has the + // new one. This is the order that matters: if we closed + // first, the capture loop's next send would race the swap + // and might see the closed publisher. + if (old != null) runCatching { old.close() } + } + + try { + // Park until [collectLatest] cancels us on the next session + // swap, OR [close] cancels [pumpJob]. The broadcaster keeps + // running through the cancellation; only close() stops it. + awaitCancellation() + } finally { + // Intentionally do NOT close the broadcaster here. + // collectLatest cancels this iteration on every session + // swap; closing the broadcaster would create the exact + // 50–150 ms gap this whole path exists to avoid. + } + } + + /** + * Legacy iteration body — used for IETF reference speakers and any + * future [NestsSpeaker] that doesn't implement + * [HotSwappablePublisherSource]. Identical to the pre-hot-swap + * behaviour: open a fresh handle on each session, close it on + * cancellation. + */ + private suspend fun runLegacyIteration(sp: NestsSpeaker) { + val handle = + try { + sp.startBroadcasting(onLevel) + } catch (ce: kotlinx.coroutines.CancellationException) { + throw ce + } catch (_: Throwable) { + null + } ?: return + if (closed) { + runCatching { handle.close() } + return + } + if (desiredMuted) { + runCatching { handle.setMuted(true) } + } + liveHandle.set(handle) + try { + awaitCancellation() + } finally { + if (liveHandle.get() === handle) liveHandle.set(null) + runCatching { handle.close() } + } + } + override suspend fun setMuted(muted: Boolean) { if (closed) return desiredMuted = muted + // Apply to whichever live underlying exists. Hot-swap and + // legacy paths are mutually exclusive in steady state — a + // moq-lite session's wrapper never has a live legacy handle, + // and vice versa — but both reads are cheap and harmless if + // the unused one is null. + hotSwapBroadcaster?.setMuted(muted) liveHandle.get()?.let { runCatching { it.setMuted(muted) } } } @@ -428,6 +599,13 @@ private class ReissuingBroadcastHandle( if (closed) return closed = true pumpJob?.cancel() + // Tear down whichever path was active. For hot-swap, stopping + // the broadcaster releases the mic + encoder AND closes the + // current publisher (broadcaster.stop calls publisher.close + // internally). For legacy, the per-session handle's close + // releases its own broadcaster. + hotSwapBroadcaster?.let { runCatching { it.stop() } } + hotSwapBroadcaster = null liveHandle.getAndSet(null)?.let { runCatching { it.close() } } onClose(this) } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt index e849d1bd3..385a68188 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt @@ -39,7 +39,7 @@ import kotlinx.coroutines.launch class NestMoqLiteBroadcaster( private val capture: AudioCapture, private val encoder: OpusEncoder, - private val publisher: MoqLitePublisherHandle, + initialPublisher: MoqLitePublisherHandle, private val scope: CoroutineScope, /** * Number of consecutive Opus frames to pack into a single moq-lite @@ -98,6 +98,26 @@ class NestMoqLiteBroadcaster( @Volatile private var muted: Boolean = false + /** + * Active publisher the capture loop pushes frames into. Mutable + + * `@Volatile` so [swapPublisher] can atomically retarget the loop + * onto a fresh moq-lite session's publisher when the + * [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker] + * orchestrator recycles the WebTransport session for a JWT + * refresh — without restarting capture or the encoder. The + * capture loop snapshots this reference once per frame, so a swap + * takes effect on the very next frame and any in-flight send on + * the previous publisher is allowed to complete (or fail + * gracefully, caught by `runCatching`). + * + * Per-instance group sequence: each publisher restarts at sequence + * 0, since the publisher state lives inside [MoqLitePublisherHandle] + * (one per moq-lite session). That's fine — listeners use group + * sequence for ordering within a single subscription's uni-stream- + * per-group flow, not across publisher cycles. + */ + @Volatile private var publisher: MoqLitePublisherHandle = initialPublisher + /** * Start capturing + encoding + publishing in the background. * Returns immediately. Calling twice is an error. If @@ -161,6 +181,16 @@ class NestMoqLiteBroadcaster( // we bail. publisher.send returning `false` (no inbound // subscriber) is NOT counted — empty rooms are normal. var consecutiveSendErrors = 0 + // Track which publisher we last sent to. On swapPublisher + // (JWT-refresh hot swap), the snapshot below picks up the + // new reference; we reset framesInCurrentGroup so the + // first frame on the new publisher starts a fresh group + // (the new publisher's group sequence is 0 anyway). Without + // this, a mid-group swap would feed the new publisher + // frames as if they continued the old publisher's group, + // and the relay would see two unrelated uni streams under + // the same logical group. + var lastPublisher: MoqLitePublisherHandle = publisher try { while (true) { val pcm = capture.readFrame() ?: break @@ -181,16 +211,28 @@ class NestMoqLiteBroadcaster( } if (opus.isEmpty()) continue if (muted) continue + // Snapshot the publisher reference once per frame. + // If [swapPublisher] mid-loop installed a new + // reference, pick it up here and reset the group + // counter so the new publisher's first frame + // starts a clean group. Frames already in flight + // on the previous publisher are allowed to + // complete (or fail gracefully on `runCatching`). + val current = publisher + if (current !== lastPublisher) { + framesInCurrentGroup = 0 + lastPublisher = current + } // Pack [framesPerGroup] consecutive Opus frames // into the same moq-lite group / QUIC uni stream // before FINning. See the [framesPerGroup] kdoc // for the production cliff this works around. val sendOutcome = runCatching { - publisher.send(opus) + current.send(opus) framesInCurrentGroup += 1 if (framesInCurrentGroup >= framesPerGroup) { - publisher.endGroup() + current.endGroup() framesInCurrentGroup = 0 } } @@ -235,11 +277,14 @@ class NestMoqLiteBroadcaster( } } // EOF on the capture side. Flush whatever's in the - // open group so the relay sees its FIN and the - // listener doesn't sit on a half-delivered group - // buffer. Mirrors the per-group endGroup above. + // open group on the most-recently-used publisher so + // the relay sees its FIN and the listener doesn't + // sit on a half-delivered group buffer. Use + // [lastPublisher] (not the live [publisher] field) + // because a swap-then-EOF would otherwise FIN a + // group on a publisher that didn't open it. if (framesInCurrentGroup > 0) { - runCatching { publisher.endGroup() } + runCatching { lastPublisher.endGroup() } } } catch (ce: CancellationException) { throw ce @@ -264,6 +309,38 @@ class NestMoqLiteBroadcaster( this.muted = muted } + /** + * Hot-swap the underlying publisher. Used by + * [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker]'s + * orchestrator to retarget the broadcaster onto a fresh moq-lite + * session's publisher when the JWT-refresh window fires — without + * tearing down the AudioRecord / Opus encoder. Returns the previous + * publisher reference so the caller can close it after the new one + * is wired up. + * + * **Closing the returned old publisher** is the caller's + * responsibility — this method only swaps the reference so the + * capture loop's next snapshot picks up the new publisher. The + * caller typically does: + * + * 1. Open a new moq-lite session, mint a new publisher on it. + * 2. Call [swapPublisher] with the new publisher; cache the old. + * 3. Close the old publisher (FINs the announce bidi + current + * group's uni stream on the old session). + * 4. Close the old moq-lite session (drops the WebTransport). + * + * No-op if the broadcaster is already [stop]ped (returns null) — the + * capture loop is already gone, so swapping in a new publisher would + * just leak it. + */ + fun swapPublisher(newPublisher: MoqLitePublisherHandle): MoqLitePublisherHandle? { + if (stopped) return null + val old = publisher + if (old === newPublisher) return null + publisher = newPublisher + return old + } + /** * Stop the loop, release the mic, release the encoder, close the * moq-lite publisher (which sends `Announce(Ended)` on every active 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 deec824af..ec39ca5ac 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt @@ -45,7 +45,27 @@ class NestPlayer( private val decoder: OpusDecoder, private val player: AudioPlayer, private val scope: CoroutineScope, + /** + * Number of decoded PCM frames to buffer before starting the underlying + * [AudioPlayer]. Without pre-roll, the device begins consuming audio the + * instant the first frame arrives and underruns at the first decode that + * misses its 20 ms cadence — the most common cause of perceptible + * dropouts on devices where Compose recomposition or GC briefly stalls + * the decode loop. + * + * Production callers typically pass `5` (≈ 100 ms) — long enough to mask + * a typical Main-thread hiccup, short enough that listeners don't notice + * the join latency. Tests pass `0` to preserve the synchronous test + * scheduler behaviour the existing assertions rely on. + * + * Default is `0` so existing tests stand without modification. + */ + private val prerollFrames: Int = 0, ) { + init { + require(prerollFrames >= 0) { "prerollFrames must be >= 0, got $prerollFrames" } + } + private var job: Job? = null private var stopped = false @@ -73,9 +93,34 @@ class NestPlayer( check(!stopped) { "NestPlayer already stopped" } check(job == null) { "NestPlayer.play already called" } - player.start() + // 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. + // + // 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. job = scope.launch { + val preroll = ArrayDeque(prerollFrames.coerceAtLeast(1)) + var started = false + + suspend fun startAndFlushIfNeeded() { + if (started) return + if (preroll.isEmpty()) return + player.start() + started = true + while (preroll.isNotEmpty()) { + player.enqueue(preroll.removeFirst()) + } + } try { objects.collect { obj -> val pcm = @@ -95,9 +140,21 @@ class NestPlayer( } if (pcm.isNotEmpty()) { onLevel(peakAmplitude(pcm)) - player.enqueue(pcm) + if (started) { + player.enqueue(pcm) + } else { + preroll.addLast(pcm) + if (preroll.size >= prerollFrames) { + startAndFlushIfNeeded() + } + } } } + // Flow ended without enough frames to fill the pre-roll + // (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() } catch (ce: CancellationException) { throw ce } catch (t: Throwable) { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index 13e4c9696..b6169fd03 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -202,8 +202,6 @@ class MoqLiteSession internal constructor( endGroup = endGroup, ) val bidi = transport.openBidiStream() - bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) - bidi.write(MoqLiteCodec.encodeSubscribe(request)) // Single long-running collector for the bidi's whole lifetime. // The collector parses the SubscribeResponse inline, signals it @@ -228,16 +226,31 @@ class MoqLiteSession internal constructor( val frames = Channel(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST) val responseDeferred = CompletableDeferred() - // 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. + // Pre-register the subscription BEFORE launching the collector + // AND before the SUBSCRIBE goes on the wire. The order matters + // for two reasons: + // + // 1. Collector race (original reason): 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. + // + // 2. First-group race (audit fix #3): the relay can open + // the first group's uni stream BEFORE our subscribe() + // continuation re-enters [state] to register `id`. If + // the SUBSCRIBE bytes hit the wire before the map entry + // lands, [drainOneGroup] looks the id up against an + // empty map and silently drops the frame. Registering + // the id before writing the SUBSCRIBE closes the + // window — by the time the relay can have parsed the + // message and opened a uni stream in response, the map + // already has our entry. val sub = ListenerSubscription( id = id, @@ -249,18 +262,44 @@ class MoqLiteSession internal constructor( subscriptionsBySubscribeId[id] = sub if (groupPump == null) groupPump = scope.launch { pumpUniStreams() } } + // Now that the subscription is registered, push the SUBSCRIBE + // bytes. If `bidi.write` throws (transport torn down, peer + // reset) we'd otherwise leave an orphaned map entry whose + // frames channel never closes — the response collector hasn't + // been launched yet so its idempotent cleanup wouldn't run. + // Roll back explicitly on throw. + try { + bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) + bidi.write(MoqLiteCodec.encodeSubscribe(request)) + } catch (t: Throwable) { + state.withLock { subscriptionsBySubscribeId.remove(id) } + frames.close() + runCatching { bidi.finish() } + throw t + } scope.launch { val responseBuffer = MoqLiteFrameBuffer() var responseParsed = false + // typeCode hoisted outside the collect lambda — `readVarint` + // advances `pos` permanently while `readSizePrefixed` rolls + // its own length-varint back on incomplete-body but does NOT + // roll back the type-code that's already been consumed. If + // the response chunks split between type and body (possible + // under MTU pressure / fragmentation), a per-tick `val + // typeCode = readVarint()` would read the body's size + // prefix as the type code on the next chunk and misframe + // the response. Mirrors the same fix in [handleInboundBidi]. + var typeCode: Long? = null try { bidi.incoming().collect { chunk -> if (!responseParsed) { responseBuffer.push(chunk) - val typeCode = responseBuffer.readVarint() ?: return@collect + if (typeCode == null) typeCode = responseBuffer.readVarint() + val tc = typeCode ?: return@collect val body = responseBuffer.readSizePrefixed() ?: return@collect val out = MoqWriter() - out.writeVarint(typeCode) + out.writeVarint(tc) out.writeVarint(body.size.toLong()) out.writeBytes(body) val resp = MoqLiteCodec.decodeSubscribeResponse(out.toByteArray())