From 72295915dee9d92a499358626694800c59116111 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 6 May 2026 20:34:39 +0000 Subject: [PATCH] fix(quic): tier-local round-robin so priority survives every drain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous priority-then-round-robin shape applied the rotating start-index globally over the sorted list, so the cross-tier order flipped on alternate drains: drain N had high-priority first, drain N+1 advanced streamRoundRobinStart and had low-priority first. The priority hint was silently defeated under any sustained traffic. Replace it with strict priority across tiers + round-robin only within each same-priority tier. Higher tiers always drain ahead of lower ones; same-priority peers continue to take turns via the existing rotating start. Default-priority callers see no behaviour change (single tier, identical rotation semantics). Tighten the test: drain twice and assert the higher-priority stream emits first on BOTH drains — the regression case that the single- drain version of the test missed. Add a regression guard for the same-priority round-robin so a future refactor can't silently serialise on the first stream. https://claude.ai/code/session_01KWdr4RjVvyYZfEuPVaQfUa --- .../quic/connection/QuicConnectionWriter.kt | 112 ++++++++++-------- .../connection/QuicConnectionWriterTest.kt | 100 +++++++++++----- 2 files changed, 132 insertions(+), 80 deletions(-) diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index ddf7dda5b..7b5598738 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -410,58 +410,74 @@ private fun buildApplicationPacket( // insertion-ordered and stays in sync with the streams map. val streamsView = conn.streamsListLocked() if (streamsView.isNotEmpty()) { - // Priority-then-round-robin: stable sortedByDescending preserves - // insertion order within a tier, so same-priority streams keep - // the rotating start-index round-robin behaviour. Higher-priority - // streams (e.g. moq-lite newer-sequence group streams) drain - // first under congestion. Default priority is 0; if every stream - // is at the default, iteration order matches pre-priority code. - // Cost: O(N log N) per drain pass and one transient list - // allocation. N is small (1–10 in the moq-lite audio path); if - // it ever grows enough to matter, switch to an indirect index - // sort or maintain an incrementally-sorted view on setPriority. + // Strict priority across tiers, round-robin within each tier. + // Higher-priority streams (e.g. moq-lite newer-sequence group + // streams) ALWAYS drain ahead of lower-priority ones; the + // rotating start-index only rotates among same-priority peers. + // This is the spec-aligned shape — applying the rotation + // globally over the sorted list would flip cross-tier order on + // alternating drains, defeating the priority hint entirely. + // + // Default priority is 0; if every stream is at the default, all + // streams form a single tier and iteration order matches the + // pre-priority round-robin behaviour exactly. + // + // Cost: O(N log N) per drain pass plus one transient sorted + // list. N is small (1–10 in the moq-lite audio path); if it + // ever grows enough to matter, switch to an indirect index sort + // or maintain an incrementally-sorted view on setPriority. val sorted = if (streamsView.size > 1) streamsView.sortedByDescending { it.priority } else streamsView - val start = conn.streamRoundRobinStart % sorted.size - for (i in sorted.indices) { - if (packetBudget <= 64) break - val stream = sorted[(start + i) % sorted.size] - val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L) - // Skip if both stream and connection have no credit; FIN-only - // (zero-byte) chunks may still go through because they don't - // consume credit. - if (streamRemaining <= 0L && connBudget <= 0L && !stream.send.finPending) continue - val effectiveCap = minOf(streamRemaining, connBudget) - val maxBytes = - minOf(packetBudget - 32, effectiveCap.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) - val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue - if (chunk.data.isNotEmpty() || chunk.fin) { - frames += - StreamFrame( - streamId = stream.streamId, - offset = chunk.offset, - data = chunk.data, - fin = chunk.fin, - explicitLength = true, - ) - // Step C of the deferred-follow-ups pass: track this - // STREAM emission so RFC 9002 retransmit can re-queue - // the byte range on loss. SendBuffer.markLost (commit B) - // moves the range from in-flight back to the retransmit - // queue, and the next takeChunk replays it. - tokens += - RecoveryToken.Stream( - streamId = stream.streamId, - offset = chunk.offset, - length = chunk.data.size.toLong(), - fin = chunk.fin, - ) - packetBudget -= chunk.data.size + 32 - connBudget -= chunk.data.size - conn.sendConnectionFlowConsumed += chunk.data.size + val rotation = conn.streamRoundRobinStart + var tierStart = 0 + outer@ while (tierStart < sorted.size) { + // Walk the contiguous run of same-priority streams. + val tierPriority = sorted[tierStart].priority + var tierEnd = tierStart + 1 + while (tierEnd < sorted.size && sorted[tierEnd].priority == tierPriority) tierEnd++ + val tierSize = tierEnd - tierStart + val tierRotation = if (tierSize > 1) rotation % tierSize else 0 + for (k in 0 until tierSize) { + if (packetBudget <= 64) break@outer + val stream = sorted[tierStart + ((tierRotation + k) % tierSize)] + val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L) + // Skip if both stream and connection have no credit; FIN-only + // (zero-byte) chunks may still go through because they don't + // consume credit. + if (streamRemaining <= 0L && connBudget <= 0L && !stream.send.finPending) continue + val effectiveCap = minOf(streamRemaining, connBudget) + val maxBytes = + minOf(packetBudget - 32, effectiveCap.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) + val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue + if (chunk.data.isNotEmpty() || chunk.fin) { + frames += + StreamFrame( + streamId = stream.streamId, + offset = chunk.offset, + data = chunk.data, + fin = chunk.fin, + explicitLength = true, + ) + // Step C of the deferred-follow-ups pass: track this + // STREAM emission so RFC 9002 retransmit can re-queue + // the byte range on loss. SendBuffer.markLost (commit B) + // moves the range from in-flight back to the retransmit + // queue, and the next takeChunk replays it. + tokens += + RecoveryToken.Stream( + streamId = stream.streamId, + offset = chunk.offset, + length = chunk.data.size.toLong(), + fin = chunk.fin, + ) + packetBudget -= chunk.data.size + 32 + connBudget -= chunk.data.size + conn.sendConnectionFlowConsumed += chunk.data.size + } } + tierStart = tierEnd } - conn.streamRoundRobinStart = (start + 1) % streamsView.size + conn.streamRoundRobinStart = (rotation + 1) % streamsView.size } if (frames.isEmpty()) return null diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt index cd11c389b..006cda755 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt @@ -112,45 +112,81 @@ class QuicConnectionWriterTest { @Test fun writer_drains_higher_priority_streams_before_lower_priority() { - // T11.3 follow-up: the writer's drain loop must iterate streams - // in descending priority order so moq-lite group streams with a - // higher sequence number drain ahead of older ones under - // congestion. Pre-fix this test, the writer iterated in stable - // round-robin order regardless of priority — so a backlog of - // retransmits on an older group could starve the listener of - // fresh frames. We pin the load-bearing invariant by inspecting - // StreamFrame order in a single emitted packet: low-priority - // stream is opened FIRST (so insertion order would normally win - // round-robin), but the high-priority stream's bytes must land - // earlier in the packet. + // T11.3 follow-up: priority must dominate iteration order on + // EVERY drain, not just the first. The naive "sort by priority + // then apply the existing rotating start globally" shape looks + // right at a glance but the rotating start advances on every + // drain, so cross-tier ordering flips on alternate drains and + // the priority hint is silently defeated. The correct shape is + // strict priority across tiers + round-robin only within a + // tier, which we pin here by draining TWICE and asserting the + // higher-priority stream's StreamFrame lands first in BOTH + // packets. Low-priority stream is opened FIRST so insertion- + // order can't accidentally pass for priority ordering. runBlocking { val (client, pipe) = connectedClient() val low = client.openBidiStream() val high = client.openBidiStream() low.priority = 0 high.priority = 10 - // Distinct payloads small enough to coexist in one packet. - val lowPayload = ByteArray(200) { 0xAA.toByte() } - val highPayload = ByteArray(200) { 0xBB.toByte() } - low.send.enqueue(lowPayload) - high.send.enqueue(highPayload) - val datagram = drainOutbound(client, nowMillis = 0L) - assertNotNull(datagram, "drain must emit a packet") - val frames = pipe.decryptClientApplicationFrames(datagram) - assertNotNull(frames, "decrypt must succeed at the application level") - val streamFrames = frames.filterIsInstance() - assertEquals( - 2, - streamFrames.size, - "expected one StreamFrame per stream in this drain, got $streamFrames", - ) - assertEquals( - high.streamId, - streamFrames[0].streamId, - "higher-priority stream must drain first; saw ${streamFrames.map { it.streamId }}", - ) - assertEquals(low.streamId, streamFrames[1].streamId) + repeat(2) { round -> + low.send.enqueue(ByteArray(200) { 0xAA.toByte() }) + high.send.enqueue(ByteArray(200) { 0xBB.toByte() }) + + val datagram = drainOutbound(client, nowMillis = 0L) + assertNotNull(datagram, "drain on round $round must emit a packet") + val frames = pipe.decryptClientApplicationFrames(datagram) + assertNotNull(frames, "decrypt must succeed on round $round") + val streamFrames = frames.filterIsInstance() + assertEquals( + 2, + streamFrames.size, + "expected one StreamFrame per stream on round $round, got $streamFrames", + ) + assertEquals( + high.streamId, + streamFrames[0].streamId, + "higher-priority stream must drain first on round $round; " + + "saw ${streamFrames.map { it.streamId }}", + ) + assertEquals(low.streamId, streamFrames[1].streamId, "low second on round $round") + } + } + } + + @Test + fun writer_round_robins_within_a_priority_tier() { + // Regression guard for the tier-local round-robin: same-priority + // streams must still rotate so an early-opened stream doesn't + // monopolise a packet's stream-frame slot indefinitely. We open + // three streams at the default (0) priority and verify the + // rotating start advances by one per drain, matching the + // pre-priority behaviour. + runBlocking { + val (client, pipe) = connectedClient() + val a = client.openBidiStream() + val b = client.openBidiStream() + val c = client.openBidiStream() + // All default priority — single tier, three streams. + val expectedRotation = + listOf( + listOf(a.streamId, b.streamId, c.streamId), + listOf(b.streamId, c.streamId, a.streamId), + listOf(c.streamId, a.streamId, b.streamId), + ) + for ((round, expected) in expectedRotation.withIndex()) { + a.send.enqueue(ByteArray(64) { 0xA1.toByte() }) + b.send.enqueue(ByteArray(64) { 0xB2.toByte() }) + c.send.enqueue(ByteArray(64) { 0xC3.toByte() }) + + val datagram = drainOutbound(client, nowMillis = 0L) + assertNotNull(datagram, "drain $round must emit a packet") + val frames = pipe.decryptClientApplicationFrames(datagram) + assertNotNull(frames, "decrypt must succeed on round $round") + val ids = frames.filterIsInstance().map { it.streamId } + assertEquals(expected, ids, "round $round round-robin order") + } } }