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)) + } +}