From f1034b1f5313c9bd6a9925c04349108111097af5 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 6 May 2026 20:25:01 +0000 Subject: [PATCH] feat(quic): priority-aware stream scheduling for moq-lite groups Bias the QUIC connection writer's drain loop toward higher-priority streams so moq-lite group streams with newer (higher) sequence numbers drain ahead of older ones under congestion. Implements the T11.3 follow-up flagged in nestsClient/plans/2026-05-06-stream-priority- followup.md (now removed). QuicStream gets a `@Volatile var priority: Int = 0`. The writer's streamsView iteration is replaced by a stable sortedByDescending pass so same-priority streams keep their existing rotating-start round robin while higher-priority tiers always drain first. WebTransportWriteStream gains a `setPriority(Int)` hook; the QUIC- backed adapter forwards to the underlying QuicStream, while the in-memory test fakes treat it as a no-op. MoqLiteSession.openGroupStream calls `uni.setPriority(sequence)` (saturating to Int.MAX_VALUE) to mirror moq-rs's `Publisher::serve_group`. Tests: a new InMemoryQuicPipe.decryptClientApplicationFrames helper walks past coalesced long-header packets to surface 1-RTT frames, which lets QuicConnectionWriterTest assert that the higher-priority stream's StreamFrame lands first inside a single drained packet. https://claude.ai/code/session_01KWdr4RjVvyYZfEuPVaQfUa --- .../2026-05-06-stream-priority-followup.md | 151 ------------------ .../nestsclient/moq/lite/MoqLiteSession.kt | 7 + .../nestsclient/transport/FakeWebTransport.kt | 4 + .../transport/WebTransportSession.kt | 16 ++ .../transport/QuicWebTransportFactory.kt | 18 +++ .../quic/connection/QuicConnectionWriter.kt | 18 ++- .../vitorpamplona/quic/stream/QuicStream.kt | 21 +++ .../quic/connection/InMemoryQuicPipe.kt | 40 +++++ .../connection/QuicConnectionWriterTest.kt | 44 +++++ 9 files changed, 165 insertions(+), 154 deletions(-) delete mode 100644 nestsClient/plans/2026-05-06-stream-priority-followup.md diff --git a/nestsClient/plans/2026-05-06-stream-priority-followup.md b/nestsClient/plans/2026-05-06-stream-priority-followup.md deleted file mode 100644 index 6e9578214..000000000 --- a/nestsClient/plans/2026-05-06-stream-priority-followup.md +++ /dev/null @@ -1,151 +0,0 @@ -# Stream priority for moq-lite group uni streams (T11.3 follow-up) - -**Status**: deferred — flagged in T11 (drop `bestEffort=true`), not landed. - -## Why - -`T11` removed `bestEffort=true` from `MoqLiteSession.openGroupStream` -because the QUIC contract it relied on (drop lost ranges silently -without `RESET_STREAM`) created undeliverable streams that wedge -peer reassembly buffers — the actual user-visible bug was a 30 s -silent dropout per lost packet on lossy networks. - -`bestEffort=true` was incidentally providing a "newer-groups-skip- -queued-retransmits" effect: the writer never queued retransmits in -the first place, so under congestion the loss budget naturally -biased toward dropping older lost ranges. With `bestEffort=true` -gone, all retransmits are now queued, and under sustained -congestion the writer drains streams in `streamRoundRobinStart` -order — which can mean the listener catches up on a stale group -when a fresh one would be more useful. - -The kixelated reference (`moq-rs`'s `Publisher::serve_group` in -`rs/moq-lite/src/lite/publisher.rs:347-406`) addresses this by -calling `stream.set_priority(priority.current())` on every group -stream and biasing the writer toward higher-priority (newer) -streams. We should do the same. - -## Target shape - -### `:quic` API - -Add a stable priority hook to `QuicStream`: - -```kotlin -class QuicStream(...) { - @Volatile - var priority: Int = 0 - // Higher drains first under contention. Default 0 = unchanged - // round-robin behaviour. -} -``` - -Expose at the WebTransport layer: - -```kotlin -interface WebTransportWriteStream { - fun setPriority(priority: Int) -} -``` - -### `QuicConnectionWriter` — the load-bearing change - -Today's send-frame loop (`QuicConnectionWriter.kt:411-416`): - -```kotlin -val streamsView = conn.streamsListLocked() -val start = conn.streamRoundRobinStart % streamsView.size -for (i in streamsView.indices) { - val stream = streamsView[(start + i) % streamsView.size] - // ... -} -``` - -Replace with priority-then-round-robin: - -```kotlin -val streamsView = conn.streamsListLocked() -val sorted = streamsView.sortedByDescending { it.priority } -val start = conn.streamRoundRobinStart % sorted.size -for (i in sorted.indices) { - val stream = sorted[(start + i) % sorted.size] - // ... -} -``` - -Sort is stable; same-priority streams retain insertion order, so the -existing round-robin behaviour holds within a priority tier. Higher- -priority streams always come first in the iteration. - -Cost: O(N log N) per drain pass, where N is active local-initiated -streams. N is small (1–10 in the moq-lite audio path); the -allocation of `sorted` per pass is the only real cost. If that -shows up in profiling, switch to `kotlin.collections.IntArray`- -backed indirect sort or maintain a priority-sorted list incrementally -on `setPriority` calls. - -### moq-lite wiring - -`MoqLiteSession.openGroupStream:1022-1037`: - -```kotlin -internal suspend fun openGroupStream( - subscribeId: Long, - sequence: Long, -): WebTransportWriteStream { - val uni = transport.openUniStream() - uni.setPriority(sequence.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) - uni.write(Varint.encode(MoqLiteDataType.Group.code)) - uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence))) - return uni -} -``` - -Newer groups have higher sequence → higher priority → drain first -under congestion. The saturation conversion handles broadcasts that -run long enough for `sequence` to exceed `Int.MAX_VALUE` (≈ 71 -years at our 1 group/sec production cadence; defensive only). - -## Test - -Add a unit test in `:quic` that builds a `QuicConnection`, opens -two streams, sets `.priority = 0` on the first and `.priority = 1` -on the second, queues bytes on both, and verifies the higher- -priority stream's bytes hit the wire first. Doesn't need to fake -real flow-control backpressure — pinning the iteration order via -the writer's emitted-frames tape is sufficient to catch the -regression case ("a later refactor accidentally re-introduces -round-robin order"). - -A second test in `:nestsClient` that opens 3 group streams and -asserts later-sequence streams drain before earlier ones is -nice-to-have but secondary; the `:quic`-level test pins the load- -bearing invariant. - -## Why deferred - -The change is small in lines but touches the QUIC writer's hot -path. Rebasing a future `:quic` retransmit / pacing change onto a -priority-sorted iteration order is doable but the diff conflicts -get noisy. Better landed as a focused PR after the current -interop-work series stabilises. - -Risk profile: -- **Bugs**: subtle starvation risk if a high-priority stream - always has `streamRemaining > 0` — round-robin tiebreaker within - a priority tier mitigates this for same-priority case, but the - cross-tier case needs deliberate thought (do we want strict - priority or weighted?). -- **Performance**: per-pass sort allocation. Negligible for N≤10 - but worth measuring if N grows. -- **Compat**: streams without an explicit priority default to 0, - matching today's behaviour. Existing tests should pass unchanged. - -## When to land - -After the catalog interop series is fully verified (production -audio with no degradation under realistic conditions), and ideally -alongside a `:quic` perf review pass that touches the same code -path. Not blocking on any audio bug today — `T11`'s reliable- -delivery fix is the actual correctness change; this is the -spec-aligned hardening. 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 3dc3825ed..944eb1ae1 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 @@ -1037,6 +1037,13 @@ class MoqLiteSession internal constructor( // 150 ms late still falls inside hang's default ~200 ms // jitter buffer) and avoids the dropout entirely. val uni = transport.openUniStream() + // Mirror moq-rs `Publisher::serve_group` + // (`rs/moq-lite/src/lite/publisher.rs`): newer groups get higher + // priority so the QUIC writer drains fresh audio first when + // retransmits queue up under loss. Saturating cast guards a + // theoretical broadcast that runs long enough for `sequence` to + // exceed Int.MAX_VALUE (≈ 71 years at 1 group/sec); defensive only. + uni.setPriority(sequence.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) 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 4c79793f2..875bcd568 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt @@ -162,6 +162,8 @@ class FakeBidiStream internal constructor( override suspend fun finish() { write.close() } + + override fun setPriority(priority: Int) = Unit } class FakeReadStream internal constructor( @@ -186,4 +188,6 @@ private class ChannelWriteStream( override suspend fun finish() { channel.close() } + + override fun setPriority(priority: Int) = Unit } 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 e0c952f39..f581ea986 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/WebTransportSession.kt @@ -112,6 +112,22 @@ interface WebTransportWriteStream { /** Half-close the write side (FIN). No further writes after this call. */ suspend fun finish() + + /** + * Hint to the transport about this stream's drain priority relative + * to other streams on the same session. Higher value drains first + * under congestion; same-priority streams keep round-robin order. + * Default 0 = unchanged round-robin behaviour. + * + * moq-lite uses this to bias the writer toward newer group streams + * (sequence-numbered, fresher audio) so a backlog of retransmits on + * an older group doesn't starve the listener of fresh frames. See + * `Publisher::serve_group` in `rs/moq-lite/src/lite/publisher.rs`. + * + * Implementations that don't model priority (e.g. the in-memory + * fake) MAY treat this as a no-op. + */ + fun setPriority(priority: Int) } /** 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 4f5fe7e59..019ef6e21 100644 --- a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt +++ b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt @@ -365,6 +365,10 @@ private class QuicBidiStreamAdapter( stream.send.finish() driver.wakeup() } + + override fun setPriority(priority: Int) { + stream.priority = priority + } } private class QuicReadStreamAdapter( @@ -392,6 +396,10 @@ private class QuicUniWriteStreamAdapter( stream.send.finish() driver.wakeup() } + + override fun setPriority(priority: Int) { + stream.priority = priority + } } /** Adapter for a WT peer-initiated uni stream whose prefix has been stripped. */ @@ -431,4 +439,14 @@ private class StrippedWtBidiStreamAdapter( ?: error("peer-initiated bidi stream has no finish — demux didn't wire one") finish() } + + /** + * No-op: peer-initiated bidi streams arrive through the demux as a + * [com.vitorpamplona.quic.webtransport.StrippedWtStream] which exposes + * only `send`/`finish` closures, not the underlying [QuicStream]. The + * moq-lite priority use case targets locally-opened uni group streams + * only, so this path doesn't need to model priority — see the + * [WebTransportWriteStream.setPriority] contract. + */ + override fun setPriority(priority: Int) = Unit } 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 96bd15f8f..ddf7dda5b 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -410,10 +410,22 @@ private fun buildApplicationPacket( // insertion-ordered and stays in sync with the streams map. val streamsView = conn.streamsListLocked() if (streamsView.isNotEmpty()) { - val start = conn.streamRoundRobinStart % streamsView.size - for (i in streamsView.indices) { + // 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. + 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 = streamsView[(start + i) % streamsView.size] + 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 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 b77095970..05317cb3b 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -47,6 +47,27 @@ class QuicStream( val send = SendBuffer(bestEffort = bestEffort) val receive = ReceiveBuffer() + /** + * Send-side scheduling priority. The connection writer's drain loop + * iterates streams by descending priority; same-priority streams keep + * their existing round-robin order. Higher value = drains first under + * congestion. Default 0 matches pre-priority round-robin behaviour for + * every existing call site. + * + * Used by moq-lite group streams: the publisher assigns each new group + * a priority equal to its sequence number so that newer groups + * (fresher audio) drain ahead of older ones when retransmits queue up + * on a lossy link. Mirrors `Publisher::serve_group` in + * `rs/moq-lite/src/lite/publisher.rs` (`stream.set_priority`). + * + * `@Volatile` because callers (e.g. moq-lite's openGroupStream) + * assign from arbitrary coroutines while the writer reads it under + * [com.vitorpamplona.quic.connection.QuicConnection.lock] during a + * drain pass. + */ + @Volatile + var priority: Int = 0 + /** * Bytes received and confirmed contiguous, exposed as a flow to the consumer. * diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt index 80749b9f9..9957825b1 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt @@ -327,6 +327,46 @@ class InMemoryQuicPipe( */ fun buildServerApplicationPacket(frames: List): ByteArray? = buildServerApplicationDatagram(frames) + /** + * Decrypt the application-level (1-RTT) packet inside a client-emitted + * datagram and return its frames in wire order. Test-only helper for + * assertions that depend on per-frame ordering inside a packet (e.g. + * stream priority scheduling). Walks past any coalesced long-header + * packets (Initial / Handshake) at the front of the datagram, since the + * client may still flush ACKs at those levels post-handshake. Returns + * null if no short-header packet is present or decryption fails. + */ + fun decryptClientApplicationFrames(datagram: ByteArray): List? { + if (datagram.isEmpty()) return null + var offset = 0 + while (offset < datagram.size) { + val first = datagram[offset].toInt() and 0xFF + if ((first and 0x80) == 0) { + val proto = serverApplicationRx ?: return null + val parsed = + ShortHeaderPacket.parseAndDecrypt( + bytes = datagram, + offset = offset, + dcidLen = serverScid.length, + aead = proto.aead, + key = proto.key, + iv = proto.iv, + hp = proto.hp, + hpKey = proto.hpKey, + largestReceivedInSpace = applicationPnSpace.largestReceived, + ) ?: return null + applicationPnSpace.observeInbound(parsed.packet.packetNumber, 0L) + return decodeFrames(parsed.packet.payload) + } + // Long header — skip past it using the encoded length field so + // we can inspect any short-header packet that was coalesced + // after it. + val peeked = LongHeaderPacket.peekHeader(datagram, offset) ?: return null + offset += peeked.totalLength + } + return null + } + private fun buildServerInitialPacket(crypto: ByteArray): ByteArray { val proto = PacketProtection( 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 d08b49ef2..cd11c389b 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriterTest.kt @@ -110,6 +110,50 @@ 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. + 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) + } + } + @Test fun writer_respects_connection_level_send_credit_cap() { // Audit-4 #9: pre-fix the writer ignored sendConnectionFlowCredit