Merge pull request #2689 from vitorpamplona/claude/review-nests-implementation-5kO48

Fix race conditions and improve resilience in MOQ-Lite subscribe/broadcast

nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusDecoder.kt

@@ -74,33 +74,47 @@ class MediaCodecOpusDecoder : OpusDecoder {
         // 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.
-        val collected = ArrayList<Short>(AudioFormat.FRAME_SIZE_SAMPLES)
-        while (true) {
+        //
+        // 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
+        var formatChangeAbsorbed = false
+        drain@ while (true) {
             val outputIndex = codec.dequeueOutputBuffer(bufferInfo, DEQUEUE_TIMEOUT_US)
             when {
                 outputIndex >= 0 -> {
                     val outputBuffer =
                         codec.getOutputBuffer(outputIndex)
-                            ?: continue
+                            ?: continue@drain
                     if (bufferInfo.size > 0) {
                         outputBuffer.position(bufferInfo.offset)
                         outputBuffer.limit(bufferInfo.offset + bufferInfo.size)
                         val shorts = outputBuffer.order(ByteOrder.nativeOrder()).asShortBuffer()
-                        val tmp = ShortArray(shorts.remaining())
-                        shorts.get(tmp)
-                        for (s in tmp) collected.add(s)
+                        val toCopy = minOf(shorts.remaining(), out.size - outPos)
+                        if (toCopy > 0) {
+                            shorts.get(out, outPos, toCopy)
+                            outPos += toCopy
+                        }
                     }
                     codec.releaseOutputBuffer(outputIndex, false)
                     if (bufferInfo.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) break
                     // No more buffered output for this packet.
                     if (bufferInfo.size == 0) break
-                    if (collected.size >= AudioFormat.FRAME_SIZE_SAMPLES) break
+                    if (outPos >= out.size) break
                 }
 
                 outputIndex == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED -> {
-                    // The format change carries no audio data; loop to read
-                    // the actual PCM frame.
-                    continue
+                    // The format change carries no audio data; absorb it
+                    // once and loop to read the actual PCM frame. Buggy
+                    // decoders that re-fire format-change without
+                    // producing output would otherwise busy-spin.
+                    if (formatChangeAbsorbed) break
+                    formatChangeAbsorbed = true
+                    continue@drain
                 }
 
                 outputIndex == MediaCodec.INFO_TRY_AGAIN_LATER -> {
@@ -112,7 +126,7 @@ class MediaCodecOpusDecoder : OpusDecoder {
                 }
             }
         }
-        return ShortArray(collected.size) { collected[it] }
+        return if (outPos == out.size) out else out.copyOf(outPos)
     }
 
     override fun release() {

nestsClient/src/androidMain/kotlin/com/vitorpamplona/nestsclient/audio/MediaCodecOpusEncoder.kt

@@ -21,7 +21,6 @@
 package com.vitorpamplona.nestsclient.audio
 
 import android.media.MediaCodec
-import android.media.MediaCodecInfo
 import android.media.MediaFormat
 import java.nio.ByteOrder
 
@@ -66,7 +65,7 @@ class MediaCodecOpusEncoder(
         val inputBuffer =
             codec.getInputBuffer(inputIndex)
                 ?: throw AudioException(
-                    AudioException.Kind.DecoderError,
+                    AudioException.Kind.EncoderError,
                     "MediaCodec returned null input buffer at index $inputIndex",
                 )
         inputBuffer.clear()
@@ -76,7 +75,12 @@ class MediaCodecOpusEncoder(
         presentationTimeUs += FRAME_DURATION_US
 
         // One PCM frame produces one Opus packet (sometimes after one warmup
-        // round). Drain the output queue once.
+        // round). Drain the output queue once. The format-change signal
+        // fires at most once on encoder startup; absorb it then re-poll,
+        // but never loop on it (some buggy encoders re-emit FORMAT_CHANGED
+        // without producing output, which would otherwise busy-spin at
+        // 100 Hz against the 10 ms dequeue timeout).
+        var formatChangeAbsorbed = false
         while (true) {
             val outputIndex = codec.dequeueOutputBuffer(bufferInfo, DEQUEUE_TIMEOUT_US)
             when {
@@ -91,6 +95,8 @@ class MediaCodecOpusEncoder(
                 }
 
                 outputIndex == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED -> {
+                    if (formatChangeAbsorbed) return ByteArray(0)
+                    formatChangeAbsorbed = true
                     continue
                 }
 
@@ -128,12 +134,11 @@ class MediaCodecOpusEncoder(
                 ).apply {
                     setInteger(MediaFormat.KEY_BIT_RATE, bitrate)
                     setInteger(MediaFormat.KEY_PCM_ENCODING, android.media.AudioFormat.ENCODING_PCM_16BIT)
-                    // Encoder-side AAC/Opus profile selection: SignalingDelaySamples
-                    // is implicit; nothing else required for mono speech.
-                    setInteger(
-                        MediaFormat.KEY_AAC_PROFILE,
-                        MediaCodecInfo.CodecProfileLevel.AACObjectLC,
-                    )
+                    // KEY_AAC_PROFILE on an audio/opus encoder is meaningless
+                    // (Opus has no AAC profile); historically it was silently
+                    // ignored, but stricter Codec2 stacks on Android 13+
+                    // reject the configure() call with IllegalArgumentException
+                    // and surface as DeviceUnavailable.
                 }
     }
 }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt

@@ -146,13 +146,29 @@ class MoqLiteNestsListener internal constructor(
                     )
                 }
             } finally {
-                runCatching { handle.close() }
+                // The flow's `finally` runs on both normal completion
+                // and cancellation — let cancel propagate after closing
+                // the announce handle.
+                try {
+                    handle.close()
+                } catch (ce: kotlinx.coroutines.CancellationException) {
+                    throw ce
+                } catch (_: Throwable) {
+                    // Best-effort.
+                }
             }
         }
 
     override suspend fun close() {
         if (state.value is NestsListenerState.Closed) return
-        runCatching { session.close() }
+        try {
+            session.close()
+        } catch (ce: kotlinx.coroutines.CancellationException) {
+            mutableState.value = NestsListenerState.Closed
+            throw ce
+        } catch (_: Throwable) {
+            // Best-effort.
+        }
         mutableState.value = NestsListenerState.Closed
     }
 

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt

@@ -76,21 +76,40 @@ class MoqLiteNestsSpeaker internal constructor(
             // Per the audio-rooms NIP draft + JS reference
             // (`@moq/publish/screen-B680RFft.js:5641`), publishers
             // claim a broadcast suffix equal to their pubkey hex.
-            val publisher =
+            val publisher = session.publish(broadcastSuffix = speakerPubkeyHex)
+            // From here on, the publisher is registered in the session
+            // and has a live announce/subscribe path. Anything that
+            // throws before we hand a working handle back to the caller
+            // must close the publisher to avoid leaking it inside the
+            // session's `activePublisher` slot — otherwise a subsequent
+            // startBroadcasting fails with "already publishing" and the
+            // publisher's announce state never tears down.
+            val broadcaster =
                 try {
-                    session.publish(broadcastSuffix = speakerPubkeyHex)
+                    NestMoqLiteBroadcaster(
+                        capture = captureFactory(),
+                        encoder = encoderFactory(),
+                        publisher = publisher,
+                        scope = scope,
+                        framesPerGroup = framesPerGroup,
+                    ).also {
+                        it.start(
+                            onTerminalFailure = {
+                                // Broadcaster bailed after sustained
+                                // publisher.send failures. Flip to
+                                // Failed so the reconnect orchestrator
+                                // sees a terminal state and recycles
+                                // the session — without this signal the
+                                // outward state stays on Broadcasting
+                                // and the room is silently mute.
+                                onBroadcastTerminalFailure()
+                            },
+                        )
+                    }
                 } catch (t: Throwable) {
+                    runCatching { publisher.close() }
                     throw t
                 }
-            val broadcaster =
-                NestMoqLiteBroadcaster(
-                    capture = captureFactory(),
-                    encoder = encoderFactory(),
-                    publisher = publisher,
-                    scope = scope,
-                    framesPerGroup = framesPerGroup,
-                )
-            broadcaster.start()
             mutableState.value =
                 NestsSpeakerState.Broadcasting(
                     room = current.room,
@@ -123,6 +142,22 @@ class MoqLiteNestsSpeaker internal constructor(
         }
     }
 
+    /**
+     * Called from the broadcaster's `onTerminalFailure` callback (off
+     * the speaker's coroutine). Transitions the speaker to `Failed` so
+     * the reconnect orchestrator (`ReconnectingNestsSpeaker`) observes
+     * a terminal state and recycles the session. No-op if the speaker
+     * is already in a terminal state.
+     */
+    private fun onBroadcastTerminalFailure() {
+        val current = mutableState.value
+        if (current is NestsSpeakerState.Failed || current is NestsSpeakerState.Closed) return
+        mutableState.value =
+            NestsSpeakerState.Failed(
+                reason = "broadcast pipeline gave up — likely transport loss",
+            )
+    }
+
     internal fun reportMuteState(muted: Boolean) {
         val current = mutableState.value
         if (current is NestsSpeakerState.Broadcasting) {
@@ -141,8 +176,25 @@ class MoqLiteNestsSpeaker internal constructor(
             activeHandle = null
             mutableState.value = NestsSpeakerState.Closed
         }
-        handle?.runCatching { close() }
-        runCatching { session.close() }
+        // Don't `runCatching { handle.close() }` — that swallows
+        // CancellationException too, breaking structured cancellation
+        // when the parent scope is cancelling teardown.
+        if (handle != null) {
+            try {
+                handle.close()
+            } catch (ce: kotlinx.coroutines.CancellationException) {
+                throw ce
+            } catch (_: Throwable) {
+                // Best-effort — already cleared activeHandle.
+            }
+        }
+        try {
+            session.close()
+        } catch (ce: kotlinx.coroutines.CancellationException) {
+            throw ce
+        } catch (_: Throwable) {
+            // Best-effort.
+        }
     }
 }
 
@@ -167,11 +219,29 @@ internal class MoqLiteBroadcastHandle(
     override suspend fun close() {
         if (closed) return
         closed = true
-        runCatching { broadcaster.stop() }
+        try {
+            broadcaster.stop()
+        } catch (ce: kotlinx.coroutines.CancellationException) {
+            // Even on cancel, run the rest of cleanup before rethrowing
+            // — broadcaster.stop already cancels its own job, so the
+            // mic + encoder + publisher are owed their close paths.
+            runCatching { publisher.close() }
+            parent.broadcastClosed(this)
+            throw ce
+        } catch (_: Throwable) {
+            // Best-effort; fall through to the defensive publisher.close.
+        }
         // broadcaster.stop() already calls publisher.close(); call again
         // defensively to make this method idempotent against partial
         // failures on the broadcaster.stop path.
-        runCatching { publisher.close() }
+        try {
+            publisher.close()
+        } catch (ce: kotlinx.coroutines.CancellationException) {
+            parent.broadcastClosed(this)
+            throw ce
+        } catch (_: Throwable) {
+            // Best-effort.
+        }
         parent.broadcastClosed(this)
     }
 }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsConnect.kt

@@ -291,10 +291,50 @@ internal fun buildRelayConnectTarget(
     token: String,
 ): Pair<String, String> {
     val (authority, _) = parseEndpoint(endpoint)
-    val path = "/" + namespace + "?jwt=" + token
+    // Percent-encode any character in [namespace] that would terminate the
+    // URL path (`?`, `#`, ` `) or otherwise break parsing — `roomId` comes
+    // from the kind-30312 `d` tag, which NIP-53 does NOT constrain to a
+    // safe charset, so a `d` tag containing `?` or `#` would otherwise
+    // truncate the path and split the JWT into the wrong slot.
+    val path = "/" + percentEncodePath(namespace) + "?jwt=" + token
     return authority to path
 }
 
+/**
+ * Percent-encode the bytes of [s] that aren't legal in an RFC 3986 URI
+ * `pchar`. We preserve `:` and `/` literally because the namespace uses
+ * them as structural separators (`nests/<kind>:<host_pubkey>:<roomId>`),
+ * and the relay compares the path against its claim root using the same
+ * canonical form. Anything else — including `?`, `#`, `&`, ` `, control
+ * bytes, and any non-ASCII — is encoded.
+ */
+private fun percentEncodePath(s: String): String {
+    val bytes = s.encodeToByteArray()
+    val out = StringBuilder(bytes.size)
+    for (raw in bytes) {
+        val b = raw.toInt() and 0xFF
+        val c = b.toChar()
+        val safe =
+            (c in 'A'..'Z') ||
+                (c in 'a'..'z') ||
+                (c in '0'..'9') ||
+                c == '-' || c == '_' || c == '.' || c == '~' ||
+                c == '!' || c == '$' || c == '\'' || c == '(' || c == ')' ||
+                c == '*' || c == '+' || c == ',' || c == ';' || c == '=' ||
+                c == ':' || c == '@' || c == '/'
+        if (safe) {
+            out.append(c)
+        } else {
+            out.append('%')
+            out.append(HEX[b ushr 4])
+            out.append(HEX[b and 0x0F])
+        }
+    }
+    return out.toString()
+}
+
+private val HEX = charArrayOf('0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F')
+
 /**
  * Split a typical nests endpoint URL such as `https://relay.example.com/moq`
  * or `https://relay.example.com:4443/api/v1/moq?room=abc` into the
@@ -322,16 +362,40 @@ internal fun parseEndpoint(endpoint: String): Pair<String, String> {
 
     require(authorityRaw.isNotEmpty()) { "endpoint must include an authority (got '$endpoint')" }
 
-    val portSep = authorityRaw.lastIndexOf(':')
     val hasUserInfo = authorityRaw.contains('@')
     require(!hasUserInfo) { "endpoint must not include userinfo (got '$endpoint')" }
 
+    // IPv6 literal authorities use `[host]:port` form (RFC 3986 §3.2.2);
+    // a naive `lastIndexOf(':')` finds a colon *inside* the address and
+    // breaks parsing. Detect the bracketed form first and split on the
+    // colon after `]`.
+    val (hostPart, portStr) =
+        if (authorityRaw.startsWith('[')) {
+            val closeBracket = authorityRaw.indexOf(']')
+            // closeBracket==1 would mean the literal "[]" — empty IPv6
+            // address. Reject so callers can't accidentally pass an
+            // unconfigured placeholder.
+            require(closeBracket > 1) { "malformed IPv6 authority: '$authorityRaw'" }
+            val host = authorityRaw.substring(0, closeBracket + 1)
+            val tail = authorityRaw.substring(closeBracket + 1)
+            when {
+                tail.isEmpty() -> host to null
+                tail.startsWith(':') -> host to tail.substring(1)
+                else -> error("malformed IPv6 authority tail '$tail' in '$endpoint'")
+            }
+        } else {
+            val portSep = authorityRaw.lastIndexOf(':')
+            if (portSep >= 0) {
+                authorityRaw.substring(0, portSep) to authorityRaw.substring(portSep + 1)
+            } else {
+                authorityRaw to null
+            }
+        }
+
     // Strip the port if it's the scheme default so the on-the-wire authority is
     // canonical.
     val authority =
-        if (portSep >= 0) {
-            val host = authorityRaw.substring(0, portSep)
-            val portStr = authorityRaw.substring(portSep + 1)
+        if (portStr != null) {
             val port = portStr.toIntOrNull() ?: error("malformed port '$portStr' in '$endpoint'")
             val defaultPort =
                 when (scheme) {
@@ -339,9 +403,9 @@ internal fun parseEndpoint(endpoint: String): Pair<String, String> {
                     "http", "ws" -> 80
                     else -> -1
                 }
-            if (port == defaultPort) host else "$host:$port"
+            if (port == defaultPort) hostPart else "$hostPart:$port"
         } else {
-            authorityRaw
+            hostPart
         }
 
     return authority to pathRaw

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsReconnectPolicy.kt

@@ -20,6 +20,8 @@
  */
 package com.vitorpamplona.nestsclient
 
+import kotlin.random.Random
+
 /**
  * Exponential-backoff settings for the reconnect path that
  * [connectNestsListener] / [connectNestsSpeaker] consult when the
@@ -31,12 +33,20 @@ package com.vitorpamplona.nestsclient
  * `maxAttempts` defaults to unbounded — a long-running room should
  * keep trying as long as the user hasn't left the screen; the
  * Composable's `DisposableEffect.onDispose` is the cancel signal.
+ *
+ * **Jitter**: when the relay restarts, every reconnecting client is
+ * mid-backoff at the same step. Without jitter they all retry on the
+ * same `1 s, 2 s, 4 s, …` schedule and hammer the relay during
+ * recovery. [jitter] applies AWS-style "equal jitter" to spread the
+ * herd: `delay ∈ [(1 - jitter) × base, base]`. 0.0 disables (used by
+ * deterministic tests); the default of 0.3 is a 30 % spread.
  */
 data class NestsReconnectPolicy(
     val initialDelayMs: Long = 1_000,
     val multiplier: Double = 2.0,
     val maxDelayMs: Long = 30_000,
     val maxAttempts: Int = Int.MAX_VALUE,
+    val jitter: Double = 0.3,
 ) {
     init {
         require(initialDelayMs > 0) { "initialDelayMs must be > 0, got $initialDelayMs" }
@@ -45,20 +55,31 @@ data class NestsReconnectPolicy(
             "maxDelayMs ($maxDelayMs) must be >= initialDelayMs ($initialDelayMs)"
         }
         require(maxAttempts >= 1) { "maxAttempts must be >= 1, got $maxAttempts" }
+        require(jitter in 0.0..1.0) { "jitter must be in [0.0, 1.0], got $jitter" }
     }
 
     /**
      * Delay for the [attempt]-th retry (1-indexed). attempt=1 →
      * [initialDelayMs]; subsequent attempts multiply by [multiplier]
      * and clamp at [maxDelayMs]. attempt < 1 returns 0.
+     *
+     * When [jitter] > 0, returns a uniformly-random value in
+     * `[(1 - jitter) × base, base]` using [random] as the source.
+     * Tests pass a deterministic source; production callers use the
+     * default [delayForAttempt] which seeds from [Random.Default].
      */
-    fun delayForAttempt(attempt: Int): Long {
+    fun delayForAttempt(
+        attempt: Int,
+        random: Random = Random.Default,
+    ): Long {
         if (attempt < 1) return 0L
         // Compute attempt-1 doublings so attempt=1 returns initial.
         var d = initialDelayMs.toDouble()
         repeat(attempt - 1) { d *= multiplier }
-        val clamped = d.coerceAtMost(maxDelayMs.toDouble())
-        return clamped.toLong()
+        val base = d.coerceAtMost(maxDelayMs.toDouble())
+        if (jitter <= 0.0) return base.toLong()
+        val low = base * (1.0 - jitter)
+        return (low + (base - low) * random.nextDouble()).toLong()
     }
 
     /** True when [attempt] has hit [maxAttempts] (the next retry is forbidden). */
@@ -66,6 +87,6 @@ data class NestsReconnectPolicy(
 
     companion object {
         /** Off-switch for callers that want first-shot-or-fail (tests, single-room demos). */
-        val NoRetry = NestsReconnectPolicy(maxAttempts = 1)
+        val NoRetry = NestsReconnectPolicy(maxAttempts = 1, jitter = 0.0)
     }
 }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt

@@ -114,8 +114,18 @@ suspend fun connectReconnectingNestsListener(
             var attempt = 0
             while (true) {
                 val listener =
-                    runCatching { openOnce() }.getOrElse {
-                        state.value = NestsListenerState.Failed("connect failed: ${it.message}", it)
+                    try {
+                        openOnce()
+                    } catch (ce: kotlinx.coroutines.CancellationException) {
+                        // Orchestrator scope was cancelled mid-connect —
+                        // propagate so the reconnect loop dies promptly.
+                        // `runCatching` would have swallowed this and
+                        // run one more loop iteration (potentially
+                        // re-opening a doomed listener) before the next
+                        // suspending call re-checked the cancel.
+                        throw ce
+                    } catch (t: Throwable) {
+                        state.value = NestsListenerState.Failed("connect failed: ${t.message}", t)
                         null
                     }
                 var refreshTriggered = false
@@ -151,7 +161,13 @@ suspend fun connectReconnectingNestsListener(
                         // listener; don't bump `attempt` (it's not a
                         // backoff event) so the next openOnce() runs
                         // immediately.
-                        runCatching { listener.close() }
+                        try {
+                            listener.close()
+                        } catch (ce: kotlinx.coroutines.CancellationException) {
+                            throw ce
+                        } catch (_: Throwable) {
+                            // Best-effort — the listener will GC either way.
+                        }
                         attempt = 0
                         refreshTriggered = true
                     } else if (terminal is NestsListenerState.Failed && !isUserCancelled(terminal)) {
@@ -325,8 +341,19 @@ private class ReconnectingHandle(
 
                     while (currentCoroutineContext().isActive) {
                         val handle =
-                            runCatching { opener(listener) }
-                                .getOrNull() ?: break
+                            try {
+                                opener(listener)
+                            } catch (ce: kotlinx.coroutines.CancellationException) {
+                                // Don't `break` on cancel — let it propagate
+                                // so the launched pumpJob actually dies on
+                                // unsubscribe / scope cancellation. The old
+                                // `runCatching` shape ate the cancel and ran
+                                // one extra iteration before the next
+                                // suspend re-checked active state.
+                                throw ce
+                            } catch (_: Throwable) {
+                                null
+                            } ?: break
                         liveHandleRef.set(handle)
                         try {
                             handle.objects.collect { frames.emit(it) }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt

@@ -142,8 +142,15 @@ suspend fun connectReconnectingNestsSpeaker(
             var attempt = 0
             while (true) {
                 val speaker =
-                    runCatching { openOnce() }.getOrElse {
-                        state.value = NestsSpeakerState.Failed("connect failed: ${it.message}", it)
+                    try {
+                        openOnce()
+                    } catch (ce: kotlinx.coroutines.CancellationException) {
+                        // Propagate so the orchestrator dies promptly on
+                        // scope cancellation; `runCatching` would have
+                        // eaten the cancel and run one more iteration.
+                        throw ce
+                    } catch (t: Throwable) {
+                        state.value = NestsSpeakerState.Failed("connect failed: ${t.message}", t)
                         null
                     }
                 var refreshTriggered = false
@@ -183,7 +190,13 @@ suspend fun connectReconnectingNestsSpeaker(
                         // speaker; don't bump `attempt` (it's not a
                         // backoff event) so the next openOnce() runs
                         // immediately.
-                        runCatching { speaker.close() }
+                        try {
+                            speaker.close()
+                        } catch (ce: kotlinx.coroutines.CancellationException) {
+                            throw ce
+                        } catch (_: Throwable) {
+                            // Best-effort.
+                        }
                         attempt = 0
                         refreshTriggered = true
                     } else if (terminal is NestsSpeakerState.Failed && !isUserCancelledSpeaker(terminal)) {
@@ -348,8 +361,17 @@ private class ReissuingBroadcastHandle(
                     }
                     if (closed) return@collectLatest
                     val handle =
-                        runCatching { sp.startBroadcasting() }
-                            .getOrNull() ?: return@collectLatest
+                        try {
+                            sp.startBroadcasting()
+                        } 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

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/Audio.kt

@@ -153,6 +153,9 @@ class AudioException(
         /** Decoder rejected an Opus packet (corrupted bytes, unsupported config). */
         DecoderError,
 
+        /** Encoder rejected a PCM frame. */
+        EncoderError,
+
         /** Audio device resource (AudioTrack/AudioRecord) couldn't be allocated. */
         DeviceUnavailable,
 

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt

@@ -65,7 +65,16 @@ class NestBroadcaster(
      * a stopped state and the exception propagates so the caller can
      * surface it to the user.
      */
-    fun start(onError: (AudioException) -> Unit = { /* swallow */ }) {
+    fun start(
+        /**
+         * See [NestMoqLiteBroadcaster.start]'s `onTerminalFailure` kdoc —
+         * fires once when the loop bails after MAX_CONSECUTIVE_SEND_ERRORS
+         * so the speaker can transition to Failed and the orchestrator
+         * can recycle.
+         */
+        onTerminalFailure: () -> Unit = { /* swallow */ },
+        onError: (AudioException) -> Unit = { /* swallow */ },
+    ) {
         check(!stopped) { "NestBroadcaster already stopped" }
         check(job == null) { "NestBroadcaster.start already called" }
 
@@ -82,6 +91,11 @@ class NestBroadcaster(
         }
         job =
             scope.launch {
+                // Consecutive publisher.send error count — see the
+                // moq-lite broadcaster's identical guard for rationale:
+                // a permanently-dead transport spins the mic + encoder
+                // forever without a threshold.
+                var consecutiveSendErrors = 0
                 try {
                     while (true) {
                         val pcm = capture.readFrame() ?: break
@@ -93,7 +107,7 @@ class NestBroadcaster(
                             } catch (t: Throwable) {
                                 onError(
                                     AudioException(
-                                        AudioException.Kind.DecoderError,
+                                        AudioException.Kind.EncoderError,
                                         "Opus encode failed for a frame",
                                         t,
                                     ),
@@ -102,9 +116,13 @@ class NestBroadcaster(
                             }
                         if (opus.isEmpty()) continue
                         if (muted) continue
-                        runCatching { publisher.send(opus) }
-                            .onFailure { t ->
+                        val sendOutcome = runCatching { publisher.send(opus) }
+                        sendOutcome
+                            .onSuccess {
+                                consecutiveSendErrors = 0
+                            }.onFailure { t ->
                                 if (t is CancellationException) throw t
+                                consecutiveSendErrors += 1
                                 // Network drop on send is recoverable — log via onError but
                                 // don't stop the loop; the next frame may go through.
                                 onError(
@@ -114,6 +132,18 @@ class NestBroadcaster(
                                         t,
                                     ),
                                 )
+                                if (consecutiveSendErrors >= MAX_CONSECUTIVE_SEND_ERRORS) {
+                                    onError(
+                                        AudioException(
+                                            AudioException.Kind.PlaybackFailed,
+                                            "broadcast pipeline gave up after " +
+                                                "$consecutiveSendErrors consecutive send failures",
+                                            t,
+                                        ),
+                                    )
+                                    runCatching { onTerminalFailure() }
+                                    return@launch
+                                }
                             }
                     }
                 } catch (ce: CancellationException) {
@@ -160,4 +190,9 @@ class NestBroadcaster(
         runCatching { encoder.release() }
         runCatching { publisher.close() }
     }
+
+    companion object {
+        /** See [NestMoqLiteBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS]. */
+        const val MAX_CONSECUTIVE_SEND_ERRORS: Int = 250
+    }
 }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt

@@ -103,8 +103,20 @@ class NestMoqLiteBroadcaster(
      * Returns immediately. Calling twice is an error. If
      * [AudioCapture.start] throws, the broadcaster is left stopped and
      * the exception propagates.
+     *
+     * [onTerminalFailure] fires exactly once when the loop bails after
+     * [MAX_CONSECUTIVE_SEND_ERRORS] consecutive `publisher.send`
+     * failures. The broadcaster has stopped capturing by the time this
+     * callback runs; the caller (typically [MoqLiteNestsSpeaker]) is
+     * expected to mark the speaker `Failed` so the reconnect
+     * orchestrator can recycle the session — without this signal the
+     * outward speaker state stays stuck on `Broadcasting` and the
+     * orchestrator never knows to act.
      */
-    fun start(onError: (AudioException) -> Unit = { /* swallow */ }) {
+    fun start(
+        onTerminalFailure: () -> Unit = { /* swallow */ },
+        onError: (AudioException) -> Unit = { /* swallow */ },
+    ) {
         check(!stopped) { "NestMoqLiteBroadcaster already stopped" }
         check(job == null) { "NestMoqLiteBroadcaster.start already called" }
 
@@ -123,6 +135,15 @@ class NestMoqLiteBroadcaster(
                 // [MoqLitePublisherHandle.send]'s "open on first frame"
                 // contract.
                 var framesInCurrentGroup = 0
+                // Consecutive publisher.send / endGroup throw count. A
+                // permanently-dead transport (session closed under us,
+                // every openUniStream rejected) keeps producing errors at
+                // capture cadence; without a guard the broadcaster would
+                // hold the mic open forever, drain battery, and spam
+                // onError. After [MAX_CONSECUTIVE_SEND_ERRORS] failures
+                // we bail. publisher.send returning `false` (no inbound
+                // subscriber) is NOT counted — empty rooms are normal.
+                var consecutiveSendErrors = 0
                 try {
                     while (true) {
                         val pcm = capture.readFrame() ?: break
@@ -134,7 +155,7 @@ class NestMoqLiteBroadcaster(
                             } catch (t: Throwable) {
                                 onError(
                                     AudioException(
-                                        AudioException.Kind.DecoderError,
+                                        AudioException.Kind.EncoderError,
                                         "Opus encode failed for a frame",
                                         t,
                                     ),
@@ -147,23 +168,48 @@ class NestMoqLiteBroadcaster(
                         // into the same moq-lite group / QUIC uni stream
                         // before FINning. See the [framesPerGroup] kdoc
                         // for the production cliff this works around.
-                        runCatching {
-                            publisher.send(opus)
-                            framesInCurrentGroup += 1
-                            if (framesInCurrentGroup >= framesPerGroup) {
-                                publisher.endGroup()
-                                framesInCurrentGroup = 0
+                        val sendOutcome =
+                            runCatching {
+                                publisher.send(opus)
+                                framesInCurrentGroup += 1
+                                if (framesInCurrentGroup >= framesPerGroup) {
+                                    publisher.endGroup()
+                                    framesInCurrentGroup = 0
+                                }
+                            }
+                        sendOutcome
+                            .onSuccess {
+                                consecutiveSendErrors = 0
+                            }.onFailure { t ->
+                                if (t is CancellationException) throw t
+                                consecutiveSendErrors += 1
+                                onError(
+                                    AudioException(
+                                        AudioException.Kind.PlaybackFailed,
+                                        "publisher.send failed",
+                                        t,
+                                    ),
+                                )
+                                if (consecutiveSendErrors >= MAX_CONSECUTIVE_SEND_ERRORS) {
+                                    onError(
+                                        AudioException(
+                                            AudioException.Kind.PlaybackFailed,
+                                            "broadcast pipeline gave up after " +
+                                                "$consecutiveSendErrors consecutive send failures",
+                                            t,
+                                        ),
+                                    )
+                                    // Surface the bail so the speaker
+                                    // can flip to Failed and the
+                                    // reconnect orchestrator recycles
+                                    // the session. Caller still owns
+                                    // [stop] — we don't release the mic
+                                    // ourselves to avoid double-stop
+                                    // races with a concurrent caller.
+                                    runCatching { onTerminalFailure() }
+                                    return@launch
+                                }
                             }
-                        }.onFailure { t ->
-                            if (t is CancellationException) throw t
-                            onError(
-                                AudioException(
-                                    AudioException.Kind.PlaybackFailed,
-                                    "publisher.send failed",
-                                    t,
-                                ),
-                            )
-                        }
                     }
                     // EOF on the capture side. Flush whatever's in the
                     // open group so the relay sees its FIN and the
@@ -222,5 +268,14 @@ class NestMoqLiteBroadcaster(
          * [framesPerGroup] kdoc for the full rationale + history.
          */
         const val DEFAULT_FRAMES_PER_GROUP: Int = 5
+
+        /**
+         * Maximum consecutive [MoqLitePublisherHandle.send] / [endGroup]
+         * exceptions before the broadcaster bails. At 50 fps, 250 frames
+         * is ≈ 5 s of solid failures — far longer than any transient
+         * relay hiccup, short enough to stop draining the mic when the
+         * transport is irrecoverably dead.
+         */
+        const val MAX_CONSECUTIVE_SEND_ERRORS: Int = 250
     }
 }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqBuffer.kt

@@ -126,7 +126,19 @@ class MoqReader(
 }
 
 /** Thrown when an encoded MoQ frame is malformed or truncated. */
-class MoqCodecException(
+open class MoqCodecException(
     message: String,
     cause: Throwable? = null,
 ) : RuntimeException(message, cause)
+
+/**
+ * Thrown when a MoQ control frame uses a type code we don't recognise but
+ * is otherwise well-framed (length-prefixed, fully buffered). Carries the
+ * full frame size so the caller's pump can skip just this message and
+ * keep parsing the next one — versus a generic [MoqCodecException] which
+ * means "this byte stream is corrupted, drop everything".
+ */
+class MoqUnknownTypeException(
+    val typeCode: Long,
+    val bytesConsumed: Int,
+) : MoqCodecException("unknown MoQ message type: 0x${typeCode.toString(16)}")

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodec.kt

@@ -39,6 +39,14 @@ import com.vitorpamplona.quic.Varint
  * `nestsClient/plans/2026-04-26-moq-lite-gap.md`.
  */
 object MoqCodec {
+    /**
+     * Upper bound on a control-message payload length. MoQ-transport control
+     * messages are small (subscribes, announces, parameter blocks) — 16 MiB
+     * is well above any legitimate body and below any value where `.toInt()`
+     * would risk overflow. Anything bigger is treated as a malformed peer.
+     */
+    const val MAX_CONTROL_PAYLOAD_LEN: Long = 16L * 1024L * 1024L
+
     /**
      * Encode a full control-stream frame (type + length + payload).
      */
@@ -65,12 +73,24 @@ object MoqCodec {
         val typeDec = Varint.decode(src, offset) ?: return null
         val lenDec = Varint.decode(src, offset + typeDec.bytesConsumed) ?: return null
         val payloadStart = offset + typeDec.bytesConsumed + lenDec.bytesConsumed
+        // Bound the payload length before .toInt() — varint can carry up to
+        // 2^62, which silently overflows to a negative or absurd payloadEnd
+        // and makes the (payloadEnd > src.size) check return null forever,
+        // wedging the parser. Reject anything larger than the cap up front.
+        if (lenDec.value < 0L || lenDec.value > MAX_CONTROL_PAYLOAD_LEN) {
+            throw MoqCodecException(
+                "MoQ control payload length out of bounds: ${lenDec.value} (max $MAX_CONTROL_PAYLOAD_LEN)",
+            )
+        }
         val payloadEnd = payloadStart + lenDec.value.toInt()
         if (payloadEnd > src.size) return null
 
         val type =
             MoqMessageType.fromCode(typeDec.value)
-                ?: throw MoqCodecException("unknown MoQ message type: 0x${typeDec.value.toString(16)}")
+                ?: throw MoqUnknownTypeException(
+                    typeCode = typeDec.value,
+                    bytesConsumed = payloadEnd - offset,
+                )
 
         val reader = MoqReader(src, payloadStart, payloadEnd)
         val message =

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/MoqSession.kt

@@ -412,9 +412,18 @@ class MoqSession private constructor(
                 val decoded =
                     try {
                         MoqCodec.decode(buffer) ?: break
+                    } catch (e: MoqUnknownTypeException) {
+                        // Forward-compat: a peer sending a draft-17 control
+                        // message we don't enumerate (FETCH, GOAWAY,
+                        // MAX_SUBSCRIBE_ID, …) shouldn't poison the rest of
+                        // the buffer. Skip just this frame and parse on.
+                        buffer = buffer.copyOfRange(e.bytesConsumed, buffer.size)
+                        continue
                     } catch (e: MoqCodecException) {
-                        // Drop the corrupted buffer; keep the pump alive so the
-                        // next valid frame can recover the session.
+                        // Genuinely-corrupted bytes — we have no idea where
+                        // the next valid frame starts. Drop the buffer; the
+                        // pump stays alive so the next valid frame can
+                        // recover the session.
                         buffer = ByteArray(0)
                         break
                     }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt

@@ -153,9 +153,16 @@ object MoqLiteCodec {
         typeCode: Long,
         body: MoqWriter,
     ): ByteArray {
-        val out = MoqWriter()
+        // Inline the size-prefix wrap so we don't allocate body→ByteArray
+        // and a separate wrapper buffer just to copy them into `out`. The
+        // earlier shape (`out.writeBytes(wrapSizePrefixed(body))`) made
+        // three ByteArrays per response — varintless small frames don't
+        // matter, but this is on the publisher's reply path and the
+        // pattern is cheap to fix.
+        val payload = body.toByteArray()
+        val out = MoqWriter(payload.size + 16)
         out.writeVarint(typeCode)
-        out.writeBytes(wrapSizePrefixed(body))
+        out.writeLengthPrefixedBytes(payload)
         return out.toByteArray()
     }
 

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFraming.kt

@@ -42,27 +42,42 @@ import kotlinx.coroutines.CancellationException
  * Not thread-safe — used from a single coroutine per stream.
  */
 class MoqLiteFrameBuffer {
+    // [buf] is the capacity-sized backing array; [size] tracks the
+    // high-water mark of valid data inside it. Decoupling capacity from
+    // length is what makes power-of-two growth actually amortise — the
+    // earlier shape (`buf = grown.copyOf(needed)`) immediately truncated
+    // back to `needed`, defeating the doubling and forcing a fresh
+    // allocation on every push.
     private var buf: ByteArray = ByteArray(0)
+    private var size: Int = 0
     private var pos: Int = 0
 
     /** Append [chunk] to the buffer, growing/compacting as needed. */
     fun push(chunk: ByteArray) {
         if (chunk.isEmpty()) return
         compact()
-        val needed = buf.size + chunk.size
-        // Power-of-two doubling matches MoqWriter's growth policy and
-        // avoids quadratic copies under bursty arrivals.
-        var newCap = if (buf.size == 0) 64 else buf.size
-        while (newCap < needed) newCap *= 2
-        val grown = ByteArray(newCap)
-        buf.copyInto(grown, 0, 0, buf.size)
-        chunk.copyInto(grown, buf.size, 0, chunk.size)
-        buf = grown.copyOf(needed)
+        val needed = size + chunk.size
+        if (needed > buf.size) {
+            // Power-of-two doubling matches MoqWriter's growth policy and
+            // avoids quadratic copies under bursty arrivals.
+            var newCap = if (buf.size == 0) 64 else buf.size
+            while (newCap < needed) newCap *= 2
+            val grown = ByteArray(newCap)
+            buf.copyInto(grown, 0, 0, size)
+            buf = grown
+        }
+        chunk.copyInto(buf, size, 0, chunk.size)
+        size = needed
     }
 
     /** Try to read one varint. Returns null if not enough bytes. */
     fun readVarint(): Long? {
+        if (pos >= size) return null
         val dec = Varint.decode(buf, pos) ?: return null
+        // Reject a varint whose declared length runs past the live region
+        // — Varint.decode only checks against [buf]'s capacity, but slack
+        // capacity is unfilled garbage, not real bytes from the peer.
+        if (pos + dec.bytesConsumed > size) return null
         pos += dec.bytesConsumed
         return dec.value
     }
@@ -79,7 +94,7 @@ class MoqLiteFrameBuffer {
         if (len < 0 || len > Int.MAX_VALUE) {
             throw MoqCodecException("absurd moq-lite size prefix: $len")
         }
-        if (pos + len.toInt() > buf.size) {
+        if (pos + len.toInt() > size) {
             // Roll the cursor back so the next call sees the same
             // varint and only commits when the whole payload arrives.
             pos = savedPos
@@ -91,16 +106,15 @@ class MoqLiteFrameBuffer {
     }
 
     /** Bytes still buffered after [pos] — exposed for diagnostic / EOF detection. */
-    val remaining: Int get() = buf.size - pos
+    val remaining: Int get() = size - pos
 
     /** Drop the consumed prefix when half the buffer is dead weight. */
     private fun compact() {
         if (pos == 0) return
-        if (pos < buf.size / 2) return
-        val live = buf.size - pos
-        val newBuf = ByteArray(live)
-        buf.copyInto(newBuf, 0, pos, buf.size)
-        buf = newBuf
+        if (pos < size / 2) return
+        val live = size - pos
+        if (live > 0) buf.copyInto(buf, 0, pos, size)
+        size = live
         pos = 0
     }
 }

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteMessages.kt

@@ -204,7 +204,17 @@ data class MoqLiteSubscribeOk(
     val maxLatencyMillis: Long,
     val startGroup: Long?,
     val endGroup: Long?,
-)
+) {
+    init {
+        // Mirror the bounds [MoqLiteSubscribe] enforces on the request side
+        // so a publisher building a malformed Ok reply fails loudly here
+        // rather than silently writing a truncated byte on the wire.
+        require(priority in 0..255) { "moq-lite priority must fit in a byte: $priority" }
+        require(maxLatencyMillis >= 0) { "maxLatencyMillis must be non-negative: $maxLatencyMillis" }
+        if (startGroup != null) require(startGroup >= 0) { "startGroup must be non-negative: $startGroup" }
+        if (endGroup != null) require(endGroup >= 0) { "endGroup must be non-negative: $endGroup" }
+    }
+}
 
 /**
  * Publisher's reject / drop reply. moq-lite has no SUBSCRIBE_ERROR

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt

@@ -25,6 +25,7 @@ import com.vitorpamplona.nestsclient.moq.MoqWriter
 import com.vitorpamplona.nestsclient.transport.WebTransportSession
 import com.vitorpamplona.quic.Varint
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.cancelAndJoin
@@ -202,37 +203,104 @@ class MoqLiteSession internal constructor(
         bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
         bidi.write(MoqLiteCodec.encodeSubscribe(request))
 
-        // Single long-running collector pump for the bidi's response
-        // side. Reads the SubscribeResponse, then keeps collecting
-        // until the peer FINs (or scope is cancelled, or
-        // handle.unsubscribe() FINs our side and the relay echoes).
-        // The flow completion IS the moq-lite-03 signal that the
-        // publisher has disconnected mid-broadcast — Lite-03 has no
-        // explicit "publisher gone" message; bidi close is it.
-        // Without this watch, the frames Channel below would never
-        // close on remote disconnect, and any consumer collecting
-        // from the wrapper-level [MoqLiteSubscribeHandle.frames]
-        // flow would sit silent indefinitely after a publisher
-        // cycle even though the relay is happy to serve a fresh
-        // subscribe under the same broadcast suffix.
+        // Single long-running collector for the bidi's whole lifetime.
+        // The collector parses the SubscribeResponse inline, signals it
+        // via [responseDeferred], then continues draining the bidi
+        // until the peer FINs (publisher disconnect under moq-lite
+        // Lite-03) or the transport tears down. On collector exit,
+        // the frames channel is closed so the consumer's
+        // `frames.consumeAsFlow()` ends naturally.
         //
-        // Why a single pump (vs separate response read + death
-        // watch): the underlying QUIC stream's `incoming` is
-        // backed by `Channel<ByteArray>.consumeAsFlow()` — which
-        // CANCELS the channel when the first collect ends. A second
-        // collect on a fresh `bidi.incoming()` Flow would see an
-        // already-cancelled channel and fire prematurely. Keeping
-        // one collect alive sidesteps that entirely.
-        // moq-lite's subscribe-response is a single size-prefixed
-        // message on the response side of the bidi. Read incoming
-        // chunks into a buffer until the buffer holds a full payload,
-        // then stop. We don't need a separate collector pump because
-        // post-Ok the bidi is idle (group data flows on its own uni
-        // streams).
-        val responseBuffer = MoqLiteFrameBuffer()
-        val responsePayload = readSubscribeResponseFromBidi(bidi.incoming(), responseBuffer, id)
-        when (val resp = MoqLiteCodec.decodeSubscribeResponse(responsePayload)) {
+        // Why one collector (vs separate response read + lifetime
+        // watch): the underlying QUIC stream's `incoming` is backed
+        // by `Channel<ByteArray>.consumeAsFlow()` which CANCELS the
+        // channel when the first collect ends. A second collect on a
+        // fresh `bidi.incoming()` Flow would see the already-cancelled
+        // channel and fire prematurely.
+        //
+        // Why we need a lifetime watch at all: pumpAnnounceWatch only
+        // closes frames when the relay actively sends Announce(Ended).
+        // A silent transport black-hole never reaches that path; without
+        // a per-subscription bidi watch, consumers would hang forever
+        // on dead UDP paths.
+        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.
+        val sub =
+            ListenerSubscription(
+                id = id,
+                request = request,
+                bidi = bidi,
+                frames = frames,
+            )
+        state.withLock {
+            subscriptionsBySubscribeId[id] = sub
+            if (groupPump == null) groupPump = scope.launch { pumpUniStreams() }
+        }
+
+        scope.launch {
+            val responseBuffer = MoqLiteFrameBuffer()
+            var responseParsed = false
+            try {
+                bidi.incoming().collect { chunk ->
+                    if (!responseParsed) {
+                        responseBuffer.push(chunk)
+                        val typeCode = responseBuffer.readVarint() ?: return@collect
+                        val body = responseBuffer.readSizePrefixed() ?: return@collect
+                        val out = MoqWriter()
+                        out.writeVarint(typeCode)
+                        out.writeVarint(body.size.toLong())
+                        out.writeBytes(body)
+                        val resp = MoqLiteCodec.decodeSubscribeResponse(out.toByteArray())
+                        responseDeferred.complete(resp)
+                        responseParsed = true
+                    }
+                    // Post-response chunks are silently discarded —
+                    // moq-lite leaves the bidi idle post-Ok. The signal
+                    // we care about is the flow's natural completion.
+                }
+            } catch (ce: CancellationException) {
+                if (!responseDeferred.isCompleted) responseDeferred.completeExceptionally(ce)
+                throw ce
+            } catch (t: Throwable) {
+                if (!responseDeferred.isCompleted) responseDeferred.completeExceptionally(t)
+            }
+            if (!responseDeferred.isCompleted) {
+                responseDeferred.completeExceptionally(
+                    MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id"),
+                )
+            }
+            // Idempotent: if subscribe() unwound on a Dropped response
+            // (or any throw from await), it already removed the
+            // subscription before throwing. Either way: remove + close.
+            val removed = state.withLock { subscriptionsBySubscribeId.remove(id) }
+            removed?.frames?.close()
+        }
+
+        val resp =
+            try {
+                responseDeferred.await()
+            } catch (t: Throwable) {
+                state.withLock { subscriptionsBySubscribeId.remove(id) }
+                frames.close()
+                runCatching { bidi.finish() }
+                throw t
+            }
+        when (resp) {
             is MoqLiteCodec.SubscribeResponse.Dropped -> {
+                state.withLock { subscriptionsBySubscribeId.remove(id) }
+                frames.close()
+                runCatching { bidi.finish() }
                 throw MoqLiteSubscribeException(
                     "publisher rejected subscribe id=$id: errorCode=${resp.drop.errorCode} " +
                         "reason='${resp.drop.reasonPhrase}'",
@@ -240,19 +308,6 @@ class MoqLiteSession internal constructor(
             }
 
             is MoqLiteCodec.SubscribeResponse.Ok -> {
-                val frames = Channel<MoqLiteFrame>(capacity = DEFAULT_FRAME_BUFFER, onBufferOverflow = BufferOverflow.DROP_OLDEST)
-                val sub =
-                    ListenerSubscription(
-                        id = id,
-                        request = request,
-                        ok = resp.ok,
-                        bidi = bidi,
-                        frames = frames,
-                    )
-                state.withLock {
-                    subscriptionsBySubscribeId[id] = sub
-                    if (groupPump == null) groupPump = scope.launch { pumpUniStreams() }
-                }
                 return MoqLiteSubscribeHandle(
                     id = id,
                     ok = resp.ok,
@@ -371,8 +426,15 @@ class MoqLiteSession internal constructor(
      */
     private suspend fun pumpUniStreams() {
         try {
-            transport.incomingUniStreams().collect { stream ->
-                scope.launch { drainOneGroup(stream) }
+            // coroutineScope binds each per-stream drain to this pump's
+            // job — when the pump is cancelled (session close), every
+            // drain is cancelled with it instead of leaking as a
+            // sibling on the outer [scope] until the transport's flow
+            // independently errors out.
+            kotlinx.coroutines.coroutineScope {
+                transport.incomingUniStreams().collect { stream ->
+                    launch { drainOneGroup(stream) }
+                }
             }
         } catch (ce: CancellationException) {
             throw ce
@@ -490,8 +552,13 @@ class MoqLiteSession internal constructor(
      */
     private suspend fun pumpInboundBidis() {
         try {
-            transport.incomingBidiStreams().collect { bidi ->
-                scope.launch { handleInboundBidi(bidi) }
+            // Bind each per-bidi handler to this pump's job (see
+            // [pumpUniStreams]'s identical comment) so they don't outlive
+            // bidiPump.cancelAndJoin() in [close].
+            kotlinx.coroutines.coroutineScope {
+                transport.incomingBidiStreams().collect { bidi ->
+                    launch { handleInboundBidi(bidi) }
+                }
             }
         } catch (ce: CancellationException) {
             throw ce
@@ -626,43 +693,6 @@ class MoqLiteSession internal constructor(
         return uni
     }
 
-    /**
-     * Read a size-prefixed payload from a bidi, seeded with whatever's
-     * already in [buffer] (the ControlType byte may have arrived with
-     * extra bytes). Used internally by the publisher inbound dispatch.
-     */
-    private suspend fun readSizePrefixedFromBidiInto(
-        incoming: kotlinx.coroutines.flow.Flow<ByteArray>,
-        buffer: MoqLiteFrameBuffer,
-    ): ByteArray {
-        // Try the buffer first — first chunk often contains the whole
-        // body since moq-lite messages are small and arrive as single
-        // QUIC sends.
-        buffer.readSizePrefixed()?.let { return it }
-        var done: ByteArray? = null
-        try {
-            incoming.collect { chunk ->
-                buffer.push(chunk)
-                buffer.readSizePrefixed()?.let {
-                    done = it
-                    throw EarlyExit
-                }
-            }
-        } catch (_: EarlyExit) {
-            // expected
-        } catch (ce: CancellationException) {
-            throw ce
-        }
-        return done
-            ?: throw MoqCodecException("incoming bidi closed before a complete size-prefixed body arrived")
-    }
-
-    private object EarlyExit : RuntimeException() {
-        private fun readResolve(): Any = EarlyExit
-
-        override fun fillInStackTrace(): Throwable = this
-    }
-
     suspend fun close() {
         if (closed) return
         closed = true
@@ -688,71 +718,9 @@ class MoqLiteSession internal constructor(
         check(!closed) { "MoqLiteSession is closed" }
     }
 
-    /**
-     * Read a moq-lite-03 SubscribeResponse off the bidi response side.
-     * The wire format is `[type_varint][body_size_varint][body_bytes]`
-     * — type lives OUTSIDE the size prefix (see
-     * `rs/moq-lite/src/lite/subscribe.rs::SubscribeResponse::encode`).
-     *
-     * Walks chunks into [buffer] until both the type discriminator and
-     * the size-prefixed body have arrived, then returns the contiguous
-     * `type+size+body` byte slab so [MoqLiteCodec.decodeSubscribeResponse]
-     * can parse it self-contained.
-     *
-     * Throws [MoqLiteSubscribeException] if the bidi closes before a
-     * full message arrives — that's the relay rejecting the subscribe
-     * with FIN instead of a SubscribeDrop reply.
-     */
-    private suspend fun readSubscribeResponseFromBidi(
-        incoming: kotlinx.coroutines.flow.Flow<ByteArray>,
-        buffer: MoqLiteFrameBuffer,
-        id: Long,
-    ): ByteArray {
-        // Snapshot the buffer's pos before each varint so we can roll
-        // back if not enough bytes have arrived yet — without this, a
-        // partial varint advances `pos` and the next chunk's bytes
-        // can't reconstitute it.
-        var typeCode: Long? = null
-        var body: ByteArray? = null
-        try {
-            // Some bytes may already be buffered (extra arrived with a
-            // prior message); try first without waiting for new chunks.
-            typeCode = buffer.readVarint()
-            if (typeCode != null) body = buffer.readSizePrefixed()
-            if (body != null) throw EarlyExit
-            incoming.collect { chunk ->
-                buffer.push(chunk)
-                if (typeCode == null) typeCode = buffer.readVarint()
-                if (typeCode != null && body == null) body = buffer.readSizePrefixed()
-                if (body != null) throw EarlyExit
-            }
-        } catch (_: EarlyExit) {
-            // expected
-        } catch (ce: CancellationException) {
-            throw ce
-        }
-        if (typeCode == null) {
-            throw MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id")
-        }
-        if (body == null) {
-            throw MoqLiteSubscribeException(
-                "subscribe stream FIN mid-body for id=$id (type=$typeCode)",
-            )
-        }
-        // Re-emit the contiguous `[type][size][body]` slab so
-        // [MoqLiteCodec.decodeSubscribeResponse] can parse it
-        // self-contained.
-        val out = MoqWriter()
-        out.writeVarint(typeCode!!)
-        out.writeVarint(body!!.size.toLong())
-        out.writeBytes(body!!)
-        return out.toByteArray()
-    }
-
     private class ListenerSubscription(
         val id: Long,
         val request: MoqLiteSubscribe,
-        val ok: MoqLiteSubscribeOk,
         val bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream,
         val frames: Channel<MoqLiteFrame>,
     )
@@ -827,8 +795,30 @@ class MoqLiteSession internal constructor(
                 if (publisherClosed) return false
                 if (inboundSubs.isEmpty()) return false
                 val group = currentGroup ?: openNextGroupLocked().also { currentGroup = it }
-                val framed = Varint.encode(payload.size.toLong()) + payload
-                runCatching { group.uni.write(framed) }
+                // Single-allocation framing: write the varint length
+                // directly into a buffer sized for `varint + payload`,
+                // then copy the payload after it. The previous shape
+                // (`Varint.encode(...) + payload`) allocated twice per
+                // Opus frame — once for the varint, once for the `+`
+                // concatenation — at 50 fps × N speakers that's measurable
+                // young-gen pressure on the audio hot path.
+                val payloadSize = payload.size
+                val varintLen = Varint.size(payloadSize.toLong())
+                val framed = ByteArray(varintLen + payloadSize)
+                Varint.writeTo(payloadSize.toLong(), framed, 0)
+                payload.copyInto(framed, varintLen)
+                val writeResult = runCatching { group.uni.write(framed) }
+                if (writeResult.isFailure) {
+                    // The uni stream errored (peer reset, transport closed,
+                    // FIN'd by removeInboundSubscription). Drop the dead
+                    // stream so the next send opens a fresh group instead of
+                    // re-trying on a corpse, and surface the failure to the
+                    // caller — the prior contract of "always true on
+                    // not-muted" silently masked publisher disconnects.
+                    runCatching { group.uni.finish() }
+                    currentGroup = null
+                    return false
+                }
             }
             return true
         }

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsConnectTest.kt

@@ -167,6 +167,41 @@ class NestsConnectTest {
         )
     }
 
+    @Test
+    fun parseEndpoint_handles_IPv6_authorities() {
+        // No port — naive `lastIndexOf(':')` would find a colon inside
+        // the address; the bracket-aware path keeps `[::1]` whole.
+        assertEquals(
+            "[::1]" to "/moq",
+            parseEndpoint("https://[::1]/moq"),
+        )
+        assertEquals(
+            "[2001:db8::1]:4443" to "/moq",
+            parseEndpoint("https://[2001:db8::1]:4443/moq"),
+        )
+        // Default-port stripping still applies on bracketed authorities.
+        assertEquals(
+            "[::1]" to "/",
+            parseEndpoint("https://[::1]:443/"),
+        )
+    }
+
+    @Test
+    fun buildRelayConnectTarget_percent_encodes_path_unsafe_chars() {
+        // A malicious or careless `d` tag containing `?`, `#`, ` ` etc.
+        // would otherwise truncate the URL path and shove the JWT into
+        // the wrong slot. Encoding keeps the namespace literal in path
+        // position; the relay URL-decodes before comparing to claim root.
+        val (authority, path) =
+            buildRelayConnectTarget(
+                endpoint = "https://relay.example.com/moq",
+                namespace = "nests/30312:0123:room?weird#frag",
+                token = "tok-abc",
+            )
+        assertEquals("relay.example.com", authority)
+        assertEquals("/nests/30312:0123:room%3Fweird%23frag?jwt=tok-abc", path)
+    }
+
     // ---------------------------------------------------------- fakes
 
     private class FakeNestsClient(

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsReconnectPolicyTest.kt

@@ -20,6 +20,7 @@
  */
 package com.vitorpamplona.nestsclient
 
+import kotlin.random.Random
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
@@ -29,13 +30,13 @@ import kotlin.test.assertTrue
 class NestsReconnectPolicyTest {
     @Test
     fun firstAttemptUsesInitialDelay() {
-        val p = NestsReconnectPolicy(initialDelayMs = 1_000, multiplier = 2.0, maxDelayMs = 30_000)
+        val p = NestsReconnectPolicy(initialDelayMs = 1_000, multiplier = 2.0, maxDelayMs = 30_000, jitter = 0.0)
         assertEquals(1_000, p.delayForAttempt(1))
     }
 
     @Test
     fun delayDoublesPerAttemptUntilMax() {
-        val p = NestsReconnectPolicy(initialDelayMs = 1_000, multiplier = 2.0, maxDelayMs = 30_000)
+        val p = NestsReconnectPolicy(initialDelayMs = 1_000, multiplier = 2.0, maxDelayMs = 30_000, jitter = 0.0)
         assertEquals(2_000, p.delayForAttempt(2))
         assertEquals(4_000, p.delayForAttempt(3))
         assertEquals(8_000, p.delayForAttempt(4))
@@ -48,7 +49,7 @@ class NestsReconnectPolicyTest {
 
     @Test
     fun nonStandardMultiplierStillRespectsCeiling() {
-        val p = NestsReconnectPolicy(initialDelayMs = 500, multiplier = 3.0, maxDelayMs = 5_000)
+        val p = NestsReconnectPolicy(initialDelayMs = 500, multiplier = 3.0, maxDelayMs = 5_000, jitter = 0.0)
         assertEquals(500, p.delayForAttempt(1))
         assertEquals(1_500, p.delayForAttempt(2))
         assertEquals(4_500, p.delayForAttempt(3))
@@ -63,6 +64,40 @@ class NestsReconnectPolicyTest {
         assertEquals(0L, p.delayForAttempt(-1))
     }
 
+    @Test
+    fun jitter_spreads_delay_into_expected_band() {
+        // Equal-jitter contract: delay is uniformly distributed on
+        // [(1 - jitter) * base, base]. With jitter=0.5 and base=1000,
+        // values must land in [500, 1000]. Sample a few times with a
+        // seeded RNG so the test is deterministic.
+        val p = NestsReconnectPolicy(initialDelayMs = 1_000, jitter = 0.5)
+        val rng = Random(0xC0FFEE)
+        repeat(50) {
+            val d = p.delayForAttempt(1, rng)
+            assertTrue(d in 500..1_000, "expected jittered delay in [500, 1000], got $d")
+        }
+    }
+
+    @Test
+    fun jitter_zero_yields_deterministic_base() {
+        val p = NestsReconnectPolicy(initialDelayMs = 1_000, jitter = 0.0)
+        // Random source is consulted but ignored when jitter == 0.
+        repeat(10) {
+            assertEquals(1_000, p.delayForAttempt(1, Random(it.toLong())))
+        }
+    }
+
+    @Test
+    fun jitter_one_can_collapse_to_zero_or_full_base() {
+        val p = NestsReconnectPolicy(initialDelayMs = 1_000, jitter = 1.0)
+        // jitter=1.0 → low bound is 0 → delay can be anywhere in [0, 1000].
+        val rng = Random(42)
+        repeat(50) {
+            val d = p.delayForAttempt(1, rng)
+            assertTrue(d in 0..1_000, "expected delay in [0, 1000], got $d")
+        }
+    }
+
     @Test
     fun isExhaustedHonoursMaxAttempts() {
         val p = NestsReconnectPolicy(maxAttempts = 3)
@@ -89,5 +124,7 @@ class NestsReconnectPolicyTest {
             NestsReconnectPolicy(initialDelayMs = 5_000, maxDelayMs = 1_000)
         }
         assertFailsWith<IllegalArgumentException> { NestsReconnectPolicy(maxAttempts = 0) }
+        assertFailsWith<IllegalArgumentException> { NestsReconnectPolicy(jitter = -0.1) }
+        assertFailsWith<IllegalArgumentException> { NestsReconnectPolicy(jitter = 1.1) }
     }
 }

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcasterTest.kt

@@ -111,6 +111,45 @@ class NestBroadcasterTest {
             broadcaster.stop()
         }
 
+    @Test
+    fun onTerminalFailure_fires_once_after_consecutive_send_failures() =
+        runTest {
+            // Drive MAX_CONSECUTIVE_SEND_ERRORS + 1 frames through a
+            // publisher that always throws. The broadcaster must:
+            //   - keep going past one failure (we already cover that)
+            //   - bail eventually
+            //   - fire onTerminalFailure exactly once
+            //   - stop pulling from capture
+            val frameCount = NestBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS + 50
+            val frames = List(frameCount) { shortArrayOf(it.toShort()) }
+            val capture = ScriptedCapture(frames)
+            val encoder = ScriptedEncoder(prefix = byteArrayOf(0xFF.toByte()))
+            val publisher = ThrowingPublisher()
+            val broadcaster = NestBroadcaster(capture, encoder, publisher, backgroundScope)
+            val errors = mutableListOf<AudioException>()
+            var terminalFailureCount = 0
+
+            broadcaster.start(
+                onTerminalFailure = { terminalFailureCount += 1 },
+                onError = { errors.add(it) },
+            )
+
+            // Wait until the broadcaster has bailed. The bail closes the
+            // launched job, but capture.awaitDrained only fires on EOF —
+            // so poll on the terminal-failure counter instead.
+            while (terminalFailureCount == 0) kotlinx.coroutines.yield()
+
+            assertEquals(1, terminalFailureCount, "onTerminalFailure should fire exactly once")
+            // We saw at least MAX_CONSECUTIVE_SEND_ERRORS failures before
+            // the bail, plus one "gave up" message. ScriptedEncoder
+            // doesn't fail, so all errors are publisher.send failures.
+            assertTrue(
+                errors.size >= NestBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS,
+                "should have seen ≥ ${NestBroadcaster.MAX_CONSECUTIVE_SEND_ERRORS} errors, got ${errors.size}",
+            )
+            broadcaster.stop()
+        }
+
     @Test
     fun stop_is_idempotent() =
         runTest {
@@ -207,4 +246,13 @@ class NestBroadcasterTest {
             closed = true
         }
     }
+
+    /** Publisher whose send() always throws — used to drive the bail path. */
+    private class ThrowingPublisher : MoqSession.TrackPublisher {
+        override val name: ByteArray = "throwing".encodeToByteArray()
+
+        override suspend fun send(payload: ByteArray): Boolean = error("publisher.send failure")
+
+        override suspend fun close() {}
+    }
 }

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/MoqCodecTest.kt

@@ -100,12 +100,18 @@ class MoqCodecTest {
     }
 
     @Test
-    fun unknown_message_type_is_rejected() {
-        // Craft a frame with message type 0xFFFF (large varint), length 0, no payload.
+    fun unknown_message_type_throws_typed_exception_with_full_frame_size() {
+        // Craft a frame with message type 0xFFFF (large varint), length 4,
+        // four bytes of payload. The pump uses bytesConsumed to skip past
+        // the unknown frame instead of dropping the whole buffer.
         val bogus = MoqWriter()
         bogus.writeVarint(0xFFFFL)
-        bogus.writeVarint(0L)
-        assertFailsWith<MoqCodecException> { MoqCodec.decode(bogus.toByteArray()) }
+        bogus.writeVarint(4L)
+        bogus.writeBytes(byteArrayOf(0x01, 0x02, 0x03, 0x04))
+        val encoded = bogus.toByteArray()
+        val ex = assertFailsWith<MoqUnknownTypeException> { MoqCodec.decode(encoded) }
+        assertEquals(0xFFFFL, ex.typeCode)
+        assertEquals(encoded.size, ex.bytesConsumed)
     }
 
     @Test

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFrameBufferTest.kt

@@ -0,0 +1,111 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.nestsclient.moq.lite
+
+import com.vitorpamplona.nestsclient.moq.MoqWriter
+import kotlin.test.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+
+class MoqLiteFrameBufferTest {
+    @Test
+    fun reads_size_prefixed_payload_assembled_from_multiple_chunks() {
+        val payload = ByteArray(7) { (0x10 + it).toByte() }
+        val full =
+            MoqWriter()
+                .also { it.writeLengthPrefixedBytes(payload) }
+                .toByteArray()
+
+        val buf = MoqLiteFrameBuffer()
+        // Drip-feed the payload one byte at a time so the buffer has to
+        // hold partial state across many push() calls — the "needs more
+        // bytes" branch must NOT advance pos when the payload is short.
+        for (i in full.indices) {
+            assertNull(buf.readSizePrefixed(), "must wait until payload complete")
+            buf.push(byteArrayOf(full[i]))
+        }
+        val readBack = buf.readSizePrefixed()
+        assertContentEquals(payload, readBack)
+    }
+
+    @Test
+    fun back_to_back_payloads_share_one_buffer() {
+        val w = MoqWriter()
+        w.writeLengthPrefixedBytes(byteArrayOf(0x01, 0x02))
+        w.writeLengthPrefixedBytes(byteArrayOf(0x03, 0x04, 0x05))
+        w.writeLengthPrefixedBytes(byteArrayOf(0x06))
+
+        val buf = MoqLiteFrameBuffer()
+        buf.push(w.toByteArray())
+        assertContentEquals(byteArrayOf(0x01, 0x02), buf.readSizePrefixed())
+        assertContentEquals(byteArrayOf(0x03, 0x04, 0x05), buf.readSizePrefixed())
+        assertContentEquals(byteArrayOf(0x06), buf.readSizePrefixed())
+        assertNull(buf.readSizePrefixed(), "no more frames")
+    }
+
+    @Test
+    fun growth_capacity_amortises_under_many_small_pushes() {
+        // Push 10 KB one byte at a time. The earlier shape allocated a
+        // fresh ByteArray per push; the size-tracking shape should
+        // double-and-keep, so total allocations are O(log N) — and the
+        // resulting buffer correctly reads back as one giant payload.
+        val payload = ByteArray(10_000) { (it and 0xFF).toByte() }
+        val framed =
+            MoqWriter()
+                .also { it.writeLengthPrefixedBytes(payload) }
+                .toByteArray()
+
+        val buf = MoqLiteFrameBuffer()
+        for (b in framed) buf.push(byteArrayOf(b))
+        assertContentEquals(payload, buf.readSizePrefixed())
+    }
+
+    @Test
+    fun compact_runs_after_consuming_half_then_pushing_more() {
+        // 1) Push frame A and read it (advances pos, leaves 0 live bytes).
+        // 2) Push frame B — compact() should fire, dropping the dead
+        //    prefix without copying garbage.
+        val w = MoqWriter()
+        w.writeLengthPrefixedBytes(ByteArray(100) { 0xAA.toByte() })
+        val buf = MoqLiteFrameBuffer()
+        buf.push(w.toByteArray())
+        assertContentEquals(ByteArray(100) { 0xAA.toByte() }, buf.readSizePrefixed())
+
+        val w2 = MoqWriter()
+        w2.writeLengthPrefixedBytes(byteArrayOf(0x55, 0x66))
+        buf.push(w2.toByteArray())
+        assertContentEquals(byteArrayOf(0x55, 0x66), buf.readSizePrefixed())
+        assertEquals(0, buf.remaining)
+    }
+
+    @Test
+    fun varint_must_not_read_into_uninitialised_capacity() {
+        // Push 1 byte that's a 2-byte-varint prefix. The buffer's
+        // underlying capacity is grown to 64+, but only 1 byte is live.
+        // readVarint must NOT consume the second byte of slack capacity
+        // as if it were data.
+        val buf = MoqLiteFrameBuffer()
+        // 0x40 → top 2 bits "01" → varint is 2 bytes long.
+        buf.push(byteArrayOf(0x40))
+        assertNull(buf.readVarint(), "incomplete varint must return null")
+    }
+}

nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt

@@ -385,6 +385,66 @@ class MoqLiteSessionTest {
             session.close()
         }
 
+    @Test
+    fun frames_flow_completes_when_peer_FINs_the_subscribe_bidi() =
+        runBlocking {
+            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val session = MoqLiteSession.client(clientSide, pumpScope)
+
+            val peerBidi =
+                async {
+                    val (bidi, _) = nextSubscribeBidi(serverSide)
+                    bidi.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L)))
+                    bidi
+                }
+
+            val handle = session.subscribe("speakerY", "audio/data")
+            val bidi = withTimeout(2_000) { peerBidi.await() }
+
+            // Peer FINs the subscribe bidi — moq-lite Lite-03's signal
+            // for "publisher gone, no more data ever". Without the
+            // per-subscription bidi-watch, this would leave the
+            // consumer-facing frames flow hanging forever; with the
+            // watch, the frames Channel closes and `toList()` returns.
+            bidi.finish()
+
+            withTimeout(2_000) { handle.frames.toList() }
+
+            session.close()
+        }
+
+    @Test
+    fun frames_flow_completes_when_peer_FINs_immediately_after_Ok() =
+        runBlocking {
+            // Race regression: an earlier shape of subscribe() pre-registered
+            // the subscription AFTER awaiting the Ok response. If the peer
+            // FIN'd between sending Ok and subscribe() resuming, the
+            // collector's exit-cleanup ran first against an empty map, and
+            // subscribe() then inserted the subscription — leaving the
+            // frames channel registered with no live collector to ever close
+            // it. This test forces the order by writing Ok and FIN before
+            // the listener has a chance to consume them.
+            val (clientSide, serverSide) = FakeWebTransport.pair()
+            val session = MoqLiteSession.client(clientSide, pumpScope)
+
+            val peerJob =
+                async {
+                    val (bidi, _) = nextSubscribeBidi(serverSide)
+                    bidi.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L)))
+                    bidi.finish()
+                }
+
+            val handle = session.subscribe("speakerZ", "audio/data")
+            withTimeout(2_000) { peerJob.await() }
+
+            // frames flow must complete naturally regardless of whether
+            // the FIN landed before or after subscribe()'s post-await
+            // registration.
+            withTimeout(2_000) { handle.frames.toList() }
+
+            session.close()
+        }
+
     @Test
     fun unsubscribe_FINs_the_subscribe_bidi() =
         runBlocking {

nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/OkHttpNestsClient.kt

@@ -22,8 +22,10 @@ package com.vitorpamplona.nestsclient
 
 import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
 import okhttp3.MediaType.Companion.toMediaType
 import okhttp3.OkHttpClient
 import okhttp3.Request
@@ -41,8 +43,16 @@ import kotlin.math.min
  * OkHttp-backed [NestsClient] used on JVM + Android. A shared [OkHttpClient]
  * can be injected so the app reuses connection pools / interceptors across
  * the process; the default constructor creates a dedicated client.
+ *
+ * [callTimeoutMs] enforces an upper bound on each `mintToken` round trip,
+ * including all transport / 429 retries. The injected [OkHttpClient] may
+ * have its own per-call/connect/read timeouts, but those don't bound the
+ * retry loop itself — without this watchdog, a stalled server can hold
+ * the reconnect orchestrator indefinitely (the orchestrator is suspended
+ * inside `connectNestsListener`'s mint step).
  */
 class OkHttpNestsClient(
+    private val callTimeoutMs: Long = DEFAULT_CALL_TIMEOUT_MS,
     private val httpClient: (String) -> OkHttpClient,
 ) : NestsClient {
     override suspend fun mintToken(
@@ -83,7 +93,17 @@ class OkHttpNestsClient(
         }
 
         return withContext(Dispatchers.IO) {
-            executeWithRetry(buildRequest, url)
+            // Hard upper bound on the entire mint round-trip (including
+            // retries) so a stalled server can't suspend the reconnect
+            // orchestrator indefinitely. The injected OkHttpClient's
+            // own callTimeout doesn't cover the retry loop.
+            val response =
+                try {
+                    withTimeout(callTimeoutMs) { executeWithRetry(buildRequest, url) }
+                } catch (e: TimeoutCancellationException) {
+                    throw NestsException("nests mint timed out after ${callTimeoutMs}ms for $url", e)
+                }
+            response
                 .use { response ->
                     val body = response.body.string()
                     if (!response.isSuccessful) {
@@ -98,6 +118,11 @@ class OkHttpNestsClient(
                         throw NestsException("Malformed nests response from $url", e)
                     } catch (e: IllegalArgumentException) {
                         throw NestsException("Malformed nests response from $url", e)
+                    } catch (e: IllegalStateException) {
+                        // kotlinx.serialization can throw IllegalStateException
+                        // on some malformed input shapes (e.g. unfinished
+                        // escapes) instead of SerializationException.
+                        throw NestsException("Malformed nests response from $url", e)
                     } catch (e: kotlinx.serialization.SerializationException) {
                         throw NestsException("Malformed nests response from $url", e)
                     }
@@ -180,10 +205,18 @@ class OkHttpNestsClient(
         throw NestsException("Failed to reach $url", transportError)
     }
 
-    private companion object {
+    companion object {
         private val JSON_MEDIA_TYPE = "application/json".toMediaType()
 
         private const val MAX_TRANSPORT_RETRIES = 2
+
+        /**
+         * Default upper bound on a full `mintToken` call, including all
+         * transport / 429 retries. Worst-case 429 backoff totals ~63 s
+         * per [MAX_RATE_LIMIT_RETRIES] kdoc; 90 s leaves headroom for
+         * one slow-responding 200 on top of that.
+         */
+        const val DEFAULT_CALL_TIMEOUT_MS: Long = 90_000L
     }
 }