diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsReconnectPolicy.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsReconnectPolicy.kt index 8947ffe46..b12560e7b 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/NestsReconnectPolicy.kt +++ b/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) } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt index 4590b2243..0d22c062e 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestBroadcaster.kt @@ -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 + } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt index c16a08a0b..dca21ea3a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt @@ -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 } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt index a21a5fe8b..3d5b6445a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteCodec.kt +++ b/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() } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFraming.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFraming.kt index 3050aecc2..a604e3139 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFraming.kt +++ b/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 } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteMessages.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteMessages.kt index 553274c94..3d8504c1a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteMessages.kt +++ b/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 diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsReconnectPolicyTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsReconnectPolicyTest.kt index b1fc3e07d..b7b3054e9 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/NestsReconnectPolicyTest.kt +++ b/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 { NestsReconnectPolicy(maxAttempts = 0) } + assertFailsWith { NestsReconnectPolicy(jitter = -0.1) } + assertFailsWith { NestsReconnectPolicy(jitter = 1.1) } } } diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFrameBufferTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteFrameBufferTest.kt new file mode 100644 index 000000000..d71db3188 --- /dev/null +++ b/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") + } +}