fix(nestsClient): perf + robustness — jitter, frame buffer, defensive bounds, send-error guard

- NestsReconnectPolicy gains a `jitter` parameter (default 0.3, AWS- style equal jitter). Without it, every client reconnecting after a relay restart retries on the identical 1s/2s/4s/… schedule and thunders the relay during recovery. delayForAttempt also accepts an explicit Random source for deterministic tests. - MoqLiteFrameBuffer decouples capacity from size so power-of-two growth actually amortises (the old shape allocated a fresh ByteArray then truncated capacity back to `needed` per chunk, defeating doubling). Adds a guard against varint reads past the live region into uninitialised slack capacity. - prefixWithType inlines the size-prefix wrap so a publisher SubscribeOk/Drop reply doesn't allocate three ByteArrays. - MoqLiteSubscribeOk gains the same `init { require(priority in 0..255) }` bounds check its sibling MoqLiteSubscribe has — a buggy caller building a malformed Ok reply now fails loudly instead of writing a truncated byte on the wire. - NestBroadcaster + NestMoqLiteBroadcaster track consecutive publisher.send exception count and bail after 250 (~5 s at 50 fps) instead of holding the mic open forever on a permanently dead transport. publisher.send returning `false` (no inbound subscriber) is NOT counted — empty rooms are a normal state. Adds 8 regression tests: - 5 in MoqLiteFrameBufferTest (multi-chunk reads, back-to-back payloads, growth amortisation, compact, varint past-live guard) - 3 in NestsReconnectPolicyTest (jitter spread band, jitter=0 determinism, jitter=1 collapses to 0..base) 223 tests pass, 0 failures.
2026-05-01 18:11:56 +00:00
parent f3a942dbd0
commit 9729f3a20c
8 changed files with 307 additions and 44 deletions
@@ -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)
    }
 }
@@ -82,6 +82,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
@@ -102,9 +107,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 +123,17 @@ class NestBroadcaster(
                                        t,
                                    ),
                                )
+                                if (consecutiveSendErrors >= MAX_CONSECUTIVE_SEND_ERRORS) {
+                                    onError(
+                                        AudioException(
+                                            AudioException.Kind.PlaybackFailed,
+                                            "broadcast pipeline gave up after " +
+                                                "$consecutiveSendErrors consecutive send failures",
+                                            t,
+                                        ),
+                                    )
+                                    return@launch
+                                }
                            }
                    }
                } catch (ce: CancellationException) {
@@ -160,4 +180,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
+    }
 }
@@ -123,6 +123,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
@@ -147,23 +156,43 @@ 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,
+                                        ),
+                                    )
+                                    // Falls through to the same post-loop flush path
+                                    // capture-EOF takes — caller still owns [stop],
+                                    // but the mic-burn loop is over.
+                                    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 +251,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
    }
 }
@@ -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()
    }

@@ -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
    }
 }
@@ -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