nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(quic): bestEffort streams + park CC plan indefinitely

Browse Source 
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
This commit is contained in:
Claude
2026-05-05 02:06:08 +00:00
parent 08bb3e14e1
commit fba0a5c952
10 changed files with 296 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt
+11 -3
View File
@@ -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
quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt
+9 -1
View File
@@ -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()
/**
quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt
+70 -23
View File
@@ -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<Range>,
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
quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt
+14 -3
View File
@@ -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
quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/SendBufferBestEffortTest.kt
+158
View File
@@ -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))
}
}