From fba0a5c952d5f08997d081b3aab5e187dd9c8771 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 5 May 2026 02:06:08 +0000 Subject: [PATCH] feat(quic): bestEffort streams + park CC plan indefinitely MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After drafting the congestion-control plan we concluded the audio-rooms workload doesn't actually need CC — speakers push ~8 KB/sec, which never fills any modern link's capacity. The one real concern that surfaced — STREAM retransmit wasting bandwidth on stale Opus frames on lossy uplinks — is much cheaper to fix directly than to bound via a 14-test CC subsystem. SendBuffer gains a `bestEffort: Boolean = false` constructor flag. When true, markLost drops the lost ranges instead of moving them to the retransmit queue and lets the underlying byte storage compact as if the bytes had been ACK'd. The FIN flag (if covered) also stays sent — best-effort skips FIN re-emission too. The peer may end up with a truncated stream; moq-lite's per-stream timeouts handle that. Plumbed through QuicStream → QuicConnection.openUniStream(bestEffort) → QuicWebTransportSessionState.openUniStream(bestEffort) → WebTransportSession.openUniStream(bestEffort). Default is false everywhere, so reliable streams (HTTP/3 control, moq-lite SUBSCRIBE bidi, etc.) keep RFC 9000 §3.5 semantics. MoqLiteSession.openGroupStream now passes `bestEffort = true` — group streams carry a single Opus packet, are real-time, and don't benefit from retransmit. Internal cleanup: `removeOverlap`'s `ackedNotLost: Boolean` parameter became `OverlapAction { ACK, RETRANSMIT, DROP }` so the third best- effort disposition has a name. Same code paths, same tests, just clearer at the call site. CC plan (quic/plans/2026-05-05-congestion-control.md) is updated to "parked indefinitely" with a note that this commit is the lighter- weight alternative that addresses the only practical concern. The plan is preserved as a reference if a future workload justifies CC. New tests: SendBufferBestEffortTest (6 cases — reliable baseline, best-effort drops, FIN drop in best-effort mode, partial overlap, idempotent stale loss, ACK path still works). https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ --- .../nestsclient/moq/lite/MoqLiteSession.kt | 9 +- .../nestsclient/transport/FakeWebTransport.kt | 4 +- .../transport/WebTransportSession.kt | 8 +- .../transport/QuicWebTransportFactory.kt | 4 +- quic/plans/2026-05-05-congestion-control.md | 15 +- .../quic/connection/QuicConnection.kt | 14 +- .../vitorpamplona/quic/stream/QuicStream.kt | 10 +- .../vitorpamplona/quic/stream/SendBuffer.kt | 93 ++++++++--- .../QuicWebTransportSessionState.kt | 17 +- .../quic/stream/SendBufferBestEffortTest.kt | 158 ++++++++++++++++++ 10 files changed, 296 insertions(+), 36 deletions(-) create mode 100644 quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/SendBufferBestEffortTest.kt diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index 1f7ab2161..13e4c9696 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -696,7 +696,14 @@ class MoqLiteSession internal constructor( subscribeId: Long, sequence: Long, ): com.vitorpamplona.nestsclient.transport.WebTransportWriteStream { - val uni = transport.openUniStream() + // Group streams carry a single Opus packet. They're real-time + // and best-effort — a STREAM frame arriving 200 ms late is + // worse than useless because the listener has already moved + // past that group's sequence number. Setting bestEffort=true + // tells the underlying QUIC SendBuffer to drop lost ranges + // instead of retransmitting them, bounding the bandwidth waste + // we'd otherwise incur on a lossy uplink. + val uni = transport.openUniStream(bestEffort = true) uni.write(Varint.encode(MoqLiteDataType.Group.code)) uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence))) return uni diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt index 3e33c3aa5..4c79793f2 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt @@ -73,7 +73,9 @@ class FakeWebTransport private constructor( * the moq-lite publisher path to push group data. The peer side * receives the new stream via [incomingUniStreams]. */ - override suspend fun openUniStream(): WebTransportWriteStream { + override suspend fun openUniStream(bestEffort: Boolean): WebTransportWriteStream { + // The fake transport ignores [bestEffort] — there's no loss to + // simulate in the in-memory channel. stateLock.withLock { check(open) { "session closed" } } val pipe = Channel(Channel.BUFFERED) outboundUniStreams.send(FakeReadStream(pipe)) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt index 32ae06d25..e0c952f39 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt @@ -55,8 +55,14 @@ interface WebTransportSession { * audio frames is pushed on a fresh uni stream that the publisher * opens — see `rs/moq-lite/src/lite/publisher.rs:338` * (`session.open_uni()`). + * + * If [bestEffort] is true, the underlying QUIC stream drops lost + * STREAM bytes instead of retransmitting them — for real-time + * audio (Opus group streams) this avoids pushing 200-ms-stale + * frames after a loss. Default false (RFC 9000 §3.5 reliable byte + * sequence). */ - suspend fun openUniStream(): WebTransportWriteStream + suspend fun openUniStream(bestEffort: Boolean = false): WebTransportWriteStream /** * Flow of inbound unidirectional streams initiated by the peer. diff --git a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt index a86563fc2..5f24c438d 100644 --- a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt +++ b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt @@ -278,8 +278,8 @@ class QuicWebTransportSession( return QuicBidiStreamAdapter(s, state.driver) } - override suspend fun openUniStream(): WebTransportWriteStream { - val s = state.openUniStream() + override suspend fun openUniStream(bestEffort: Boolean): WebTransportWriteStream { + val s = state.openUniStream(bestEffort = bestEffort) return QuicUniWriteStreamAdapter(s, state.driver) } diff --git a/quic/plans/2026-05-05-congestion-control.md b/quic/plans/2026-05-05-congestion-control.md index 5343b8539..7c9bbd1aa 100644 --- a/quic/plans/2026-05-05-congestion-control.md +++ b/quic/plans/2026-05-05-congestion-control.md @@ -1,6 +1,19 @@ # Congestion control for `:quic` — implementation plan -**Status:** plan, not started. +**Status:** **parked indefinitely 2026-05-05.** After drafting this +plan we concluded the audio-rooms workload doesn't actually need CC — +see [Why](#why-and-why-this-is-honestly-low-priority) below. The one +real concern that surfaced (STREAM retransmit wasting bandwidth on +stale Opus frames) is addressed by a much smaller fix — a +`bestEffort` flag on `SendBuffer` that drops lost ranges instead of +retransmitting them, set by moq-lite for group streams. That +follow-up landed on the same branch. + +This plan is preserved as a reference if a future workload (large +file transfer over `:quic`, multiple concurrent media streams, +running on heavily-shared mobile uplinks with hostile routers) ever +makes CC necessary. The architecture below is sound; we just don't +have a problem big enough to justify the implementation cost today. ## Why (and why this is honestly low-priority) diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index c3d3ade09..cbdbe2e9d 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -477,8 +477,16 @@ class QuicConnection( stream } - /** Allocate a new client-initiated unidirectional (write-only) stream. Locked. */ - suspend fun openUniStream(): QuicStream = + /** + * Allocate a new client-initiated unidirectional (write-only) stream. + * Locked. + * + * If [bestEffort] is true, the stream's [SendBuffer] drops lost + * ranges instead of retransmitting them (see + * [QuicStream.bestEffort]). Used for moq-lite group streams + * carrying real-time Opus audio. + */ + suspend fun openUniStream(bestEffort: Boolean = false): QuicStream = lock.withLock { if (nextLocalUniIndex >= peerMaxStreamsUni) { throw QuicStreamLimitException( @@ -487,7 +495,7 @@ class QuicConnection( ) } val id = StreamId.build(StreamId.Kind.CLIENT_UNI, nextLocalUniIndex++) - val stream = QuicStream(id, QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE) + val stream = QuicStream(id, QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE, bestEffort = bestEffort) stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni stream.receiveLimit = 0L // can't receive streams[id] = stream diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt index 1297250d8..b77095970 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -33,10 +33,18 @@ import kotlinx.coroutines.flow.consumeAsFlow class QuicStream( val streamId: Long, val direction: Direction, + /** + * If true, lost STREAM bytes are dropped instead of retransmitted + * (see [SendBuffer.bestEffort]). Used by moq-lite group streams + * carrying real-time Opus audio: a STREAM frame arriving 200 ms + * late is worse than useless. Default false (RFC 9000 §3.5 + * reliable byte sequence). + */ + val bestEffort: Boolean = false, ) { enum class Direction { BIDIRECTIONAL, UNIDIRECTIONAL_LOCAL_TO_REMOTE, UNIDIRECTIONAL_REMOTE_TO_LOCAL } - val send = SendBuffer() + val send = SendBuffer(bestEffort = bestEffort) val receive = ReceiveBuffer() /** diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt index 1c78e6eac..2b9d7807d 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt @@ -73,8 +73,22 @@ package com.vitorpamplona.quic.stream * drained). [markAcked] / [markLost] respect FIN — a lost range that * carried FIN is re-sent with FIN set, and the buffer's "FIN delivered" * latch ([finAcked]) only flips when the FIN-carrying range is ACK'd. + * + * # Best-effort streams + * + * Setting [bestEffort] = true changes [markLost] semantics: lost + * ranges are dropped without being re-queued for retransmit, the FIN + * bit (if present in a lost range) is not re-emitted, and the + * underlying byte storage compacts as if the range had been ACK'd. + * This is the moq-lite group-stream case — Opus audio frames + * arriving 200 ms late are worse than useless. The peer will see a + * truncated stream; that's expected and the moq-lite layer relies on + * its own per-stream timeouts. Default is false (reliable per RFC + * 9000 §3.5). */ -class SendBuffer { +class SendBuffer( + val bestEffort: Boolean = false, +) { /** * Contiguous byte storage covering `[flushedFloor, nextOffset)`. * Indexing: byte at logical offset `o` lives at `data[(o - flushedFloor).toInt()]`. @@ -251,7 +265,7 @@ class SendBuffer { length: Long, ) { synchronized(this) { - removeOverlap(inFlight, offset, length, ackedNotLost = true) + removeOverlap(inFlight, offset, length, OverlapAction.ACK) advanceFlushedFloorIfPossible() } } @@ -280,27 +294,48 @@ class SendBuffer { // to retransmit. Clamp the requested range to the retained // window to keep the operation idempotent. if (offset + length <= flushedFloor) { - if (fin && !_finAcked) _finSent = false + if (fin && !_finAcked && !bestEffort) _finSent = false return } val clampedOffset = maxOf(offset, flushedFloor) val clampedLength = (offset + length) - clampedOffset - removeOverlap(inFlight, clampedOffset, clampedLength, ackedNotLost = false) - if (fin && !_finAcked) _finSent = false + if (bestEffort) { + // Drop without retransmit (see class kdoc). Bytes vanish + // from inFlight and the data buffer can compact, the + // same way an ACK would. The FIN flag intentionally + // stays as `_finSent = true` — best-effort means we + // don't try to re-emit FIN either; the peer either saw + // it on the original wire or it's lost forever. + removeOverlap(inFlight, clampedOffset, clampedLength, OverlapAction.DROP) + advanceFlushedFloorIfPossible() + } else { + removeOverlap(inFlight, clampedOffset, clampedLength, OverlapAction.RETRANSMIT) + if (fin && !_finAcked) _finSent = false + } } } + /** + * Disposition for a range that overlapped a [markAcked] / [markLost] + * range. ACK drops the range and latches `_finAcked` if the FIN + * was covered. RETRANSMIT moves the range to the retransmit queue + * for re-emission. DROP just removes the range, used by best-effort + * streams where retransmit would deliver stale data. + */ + private enum class OverlapAction { ACK, RETRANSMIT, DROP } + /** * Walk [list] for any range overlapping `[offset, offset + length)`, - * remove the overlapping portion, and either drop it (ACK path) or - * push it onto [retransmit] (loss path). Splits ranges where the - * overlap is partial. + * remove the overlapping portion, and dispose of it per [action]: + * ACK latches `_finAcked` if FIN was covered; RETRANSMIT moves the + * covered range to the retransmit queue; DROP just removes it. + * Splits ranges where the overlap is partial. */ private fun removeOverlap( list: ArrayDeque, offset: Long, length: Long, - ackedNotLost: Boolean, + action: OverlapAction, ) { // length == 0 only meaningful for FIN-only ranges; handle by // matching the exact-offset zero-length range. @@ -315,10 +350,16 @@ class SendBuffer { val r = list[startIndex] if (r.offset == offset && r.length == 0L) { list.removeAt(startIndex) - if (ackedNotLost) { - if (r.fin) _finAcked = true - } else { - retransmit.addLast(r) + when (action) { + OverlapAction.ACK -> { + if (r.fin) _finAcked = true + } + + OverlapAction.RETRANSMIT -> { + retransmit.addLast(r) + } + + OverlapAction.DROP -> {} // discard } } } @@ -380,16 +421,22 @@ class SendBuffer { } if (coveredLen > 0L || r.length == 0L) { val coveredFin = r.fin && coveredEnd == rEnd - if (ackedNotLost) { - if (coveredFin) _finAcked = true - } else { - retransmit.addLast( - Range( - offset = coveredStart, - length = coveredLen, - fin = coveredFin, - ), - ) + when (action) { + OverlapAction.ACK -> { + if (coveredFin) _finAcked = true + } + + OverlapAction.RETRANSMIT -> { + retransmit.addLast( + Range( + offset = coveredStart, + length = coveredLen, + fin = coveredFin, + ), + ) + } + + OverlapAction.DROP -> {} // discard the covered piece } } endIndex += 1 diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt index 7ae02f24d..e9eac528c 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt @@ -173,9 +173,20 @@ class QuicWebTransportSessionState( return s } - /** Open a new client-initiated unidirectional WebTransport stream. */ - suspend fun openUniStream(): QuicStream { - val s = connection.openUniStream() + /** + * Open a new client-initiated unidirectional WebTransport stream. + * + * If [bestEffort] is true, the underlying QUIC stream's send buffer + * drops lost ranges instead of retransmitting (see + * [com.vitorpamplona.quic.stream.QuicStream.bestEffort]). The + * WT_UNI_STREAM prefix bytes are themselves enqueued onto that + * buffer, so a lost prefix-only packet won't be retransmitted + * either — fine for the moq-lite group-stream case where the + * whole stream is ephemeral, since the peer would treat it as + * truncated either way. + */ + suspend fun openUniStream(bestEffort: Boolean = false): QuicStream { + val s = connection.openUniStream(bestEffort = bestEffort) s.send.enqueue(encodeWtUniStreamPrefix(connectStreamId)) driver.wakeup() return s diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/SendBufferBestEffortTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/SendBufferBestEffortTest.kt new file mode 100644 index 000000000..034be4aaf --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/SendBufferBestEffortTest.kt @@ -0,0 +1,158 @@ +/* + * 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.quic.stream + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Best-effort SendBuffer: lost ranges are dropped instead of being + * re-queued for retransmit. The moq-lite group-stream case — Opus + * audio frames arriving 200 ms late are worse than useless. We don't + * want to bound this behavior with congestion control because the + * audio workload is too small to warrant a CC implementation, but we + * also don't want STREAM retransmit to spam stale audio onto a lossy + * uplink. + */ +class SendBufferBestEffortTest { + @Test + fun normalBuffer_lostRangeIsRequeued() { + // Sanity baseline: the default buffer (bestEffort = false) + // re-queues lost bytes onto the retransmit FIFO so the next + // takeChunk re-emits them. + val buf = SendBuffer() + buf.enqueue(byteArrayOf(1, 2, 3, 4, 5)) + val first = buf.takeChunk(maxBytes = 5) + assertNotNull(first) + assertEquals(5, first.data.size) + + buf.markLost(offset = 0L, length = 5L, fin = false) + val replay = buf.takeChunk(maxBytes = 5) + assertNotNull(replay, "lost bytes must re-emit on a reliable buffer") + assertEquals(0L, replay.offset) + assertEquals(5, replay.data.size) + } + + @Test + fun bestEffort_lostRangeIsDropped() { + // bestEffort buffer: lost bytes vanish — no retransmit, nothing + // for takeChunk to surface, the underlying byte storage + // releases as if the bytes had been ACK'd. + val buf = SendBuffer(bestEffort = true) + buf.enqueue(byteArrayOf(10, 20, 30, 40)) + val sent = buf.takeChunk(maxBytes = 4) + assertNotNull(sent) + assertEquals(4, sent.data.size) + assertEquals(4L, buf.sentOffset) + + buf.markLost(offset = 0L, length = 4L, fin = false) + + // No fresh bytes, no retransmit-queued bytes — takeChunk + // returns null. + assertNull(buf.takeChunk(maxBytes = 4), "best-effort: lost bytes are not re-emitted") + // No more readable bytes pending either. + assertEquals(0, buf.readableBytes) + } + + @Test + fun bestEffort_lostFin_isDropped_finSentRemainsTrue() { + // FIN handling: in best-effort mode we don't retransmit FIN + // either. _finSent stays true so the writer doesn't try to + // re-emit the FIN-only chunk on the next drain. The peer's + // stream may stay open from QUIC's view forever — that's the + // cost of best-effort, and moq-lite's per-stream timeouts + // handle it at the application layer. + val buf = SendBuffer(bestEffort = true) + buf.enqueue(byteArrayOf(1, 2, 3)) + buf.finish() + val withFin = buf.takeChunk(maxBytes = 3) + assertNotNull(withFin) + assertTrue(withFin.fin, "FIN attached to last chunk") + assertTrue(buf.finSent) + + buf.markLost(offset = 0L, length = 3L, fin = true) + // _finSent should NOT flip back to false — best-effort means + // we don't retransmit the FIN. + assertTrue(buf.finSent, "best-effort: lost FIN is not resurrected for retransmit") + // No retransmit pending. + assertNull(buf.takeChunk(maxBytes = 8)) + } + + @Test + fun bestEffort_partialOverlapLoss_dropsPartialButKeepsRest() { + // Edge case: loss notification covers a partial overlap of an + // in-flight range. The non-overlapped portion stays in + // inFlight (so a later ACK / loss for that portion behaves + // correctly); the overlapped piece is dropped. + val buf = SendBuffer(bestEffort = true) + buf.enqueue(byteArrayOf(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)) + val sent = buf.takeChunk(maxBytes = 10) + assertNotNull(sent) + + // Lose the middle 4 bytes [3, 7). + buf.markLost(offset = 3L, length = 4L, fin = false) + + // The dropped range vanishes; the kept-pieces around it stay + // in inFlight. takeChunk returns null because there's no + // retransmit queue and no fresh bytes. + assertNull(buf.takeChunk(maxBytes = 10)) + + // ACK the kept-pieces explicitly to verify they're tracked + // correctly. Acking [0,3) should latch flushedFloor to 3 + // (since [3,7) is dropped — gone) only after the kept-piece + // [7,10) is also released. + buf.markAcked(offset = 0L, length = 3L) + // The dropped [3,7) range was already "released" by the + // best-effort drop, so [0,3) ACK + dropped [3,7) compact the + // floor to 7. + buf.markAcked(offset = 7L, length = 3L) + // Now everything is gone. + assertNull(buf.takeChunk(maxBytes = 10)) + } + + @Test + fun bestEffort_repeatedLossIsIdempotent() { + // Defensive: a stale loss notification arriving after the + // bytes are already dropped should not throw or double-count. + val buf = SendBuffer(bestEffort = true) + buf.enqueue(byteArrayOf(1, 2, 3)) + buf.takeChunk(maxBytes = 3) + buf.markLost(offset = 0L, length = 3L, fin = false) + buf.markLost(offset = 0L, length = 3L, fin = false) // already gone + assertNull(buf.takeChunk(maxBytes = 3)) + } + + @Test + fun bestEffort_ackPathStillWorksNormally() { + // bestEffort only changes loss behavior; ACK semantics are + // unchanged. Sanity check that we didn't regress markAcked. + val buf = SendBuffer(bestEffort = true) + buf.enqueue(byteArrayOf(1, 2, 3, 4)) + val sent = buf.takeChunk(maxBytes = 4) + assertNotNull(sent) + buf.markAcked(offset = 0L, length = 4L) + // Same as a reliable buffer: acked bytes release. + assertNull(buf.takeChunk(maxBytes = 4)) + } +}