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

fix(nests): close 4 audio-dropout sources across listener + speaker

Browse Source 
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.
This commit is contained in:
Claude
2026-05-05 03:06:40 +00:00
parent 1acd54eada
commit 076b301d84
7 changed files with 618 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt
+24 -1
View File
@@ -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
nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/AudioTrackPlayer.kt
+76 -8
View File
@@ -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) {
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt
+56 -4
View File
@@ -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<NestsSpeakerState> = 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()
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt
+266 -88
View File
@@ -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<NestsSpeaker?>,
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<NestsSpeakerState> = 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<NestsSpeaker?>,
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<BroadcastHandle?>(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)
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt
+84 -7
View File
@@ -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
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestPlayer.kt
+59 -2
View File
@@ -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<ShortArray>(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) {
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
+53 -14
View File
@@ -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<MoqLiteFrame>(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST)
val responseDeferred = CompletableDeferred<MoqLiteCodec.SubscribeResponse>()
// 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())