From 0ced269b279e9cc9615cdfbd67669c4dd306750d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 4 May 2026 22:45:10 +0000 Subject: [PATCH] =?UTF-8?q?feat(quic):=20step=203=20of=20RFC=209002=20retr?= =?UTF-8?q?ansmit=20=E2=80=94=20drain=20SentPacket=20on=20ACK?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit QuicConnectionParser's AckFrame handler now drains state.sentPackets of every entry whose packet number is covered by the ACK's ranges. The drained SentPackets are returned but discarded for now; step 5 will route them to loss detection / RTT. New helper file `connection/recovery/AckedPackets.kt`: - forEachAckedPacketNumber(ack, block): inline iterator over an AckFrame's ranges in RFC 9000 §19.3.1 order. Walks first range [largestAcked - firstAckRange, largestAcked] then each additional range with `nextLargest = previousSmallest - gap - 2`, `nextSmallest = nextLargest - ackRangeLength`. Defensive clamp at PN 0 against malformed peer ACKs. - drainAckedSentPackets(sentPackets, ack): walks via forEachAckedPacketNumber and removes each matching entry from the map. Returns the drained list. Wired into QuicConnectionParser.kt:165 alongside the existing ackTracker.purgeBelow call. Tests added (9, all pass): - simpleRange / multipleRanges / singlePacketAck: range walking for typical ACK shapes - ackForUnsentPn_isNoOp / emptyMap_returnsEmptyDrain: defensive paths - ackBoundary_pn0Inclusive: PN 0 is correctly included, no underflow - forEachAckedPacketNumber_iteratesDescending / forEachAckedPacketNumber_acrossMultipleRanges_descending: iterator semantics - returnedDrain_preservesTokens: drained SentPacket retains its full tokens list — step 5+ will dispatch these to RTT / loss Mirror of neqo's `recovery/mod.rs::remove_acked` (one of the 20 recovery tests we owe per `quic/plans/2026-05-04-control-frame-retransmit.md`). The remaining loss-detection tests land in step 5. Full :quic test suite + nestsClient moq-lite tests pass. https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ --- .../quic/connection/QuicConnectionParser.kt | 17 +- .../quic/connection/recovery/AckedPackets.kt | 84 +++++++ .../connection/recovery/AckedPacketsTest.kt | 238 ++++++++++++++++++ 3 files changed, 334 insertions(+), 5 deletions(-) create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPackets.kt create mode 100644 quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPacketsTest.kt diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt index abd87f92b..d81cee1bb 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -21,6 +21,7 @@ package com.vitorpamplona.quic.connection import com.vitorpamplona.quic.QuicCodecException +import com.vitorpamplona.quic.connection.recovery.drainAckedSentPackets import com.vitorpamplona.quic.frame.AckFrame import com.vitorpamplona.quic.frame.ConnectionCloseFrame import com.vitorpamplona.quic.frame.CryptoFrame @@ -163,12 +164,18 @@ private fun dispatchFrames( for (frame in frames) { when (frame) { is AckFrame -> { - // We don't currently retransmit, so we just absorb ACKs. But - // we DO purge our own ACK tracker below the peer's largest - // acknowledged: the peer has confirmed receipt of those ACKs, - // so we don't need to keep advertising them — without this - // the range list grows unboundedly on long connections. + // Purge our own ACK tracker below the peer's largest + // acknowledged: the peer has confirmed receipt of those + // ACKs, so we don't need to keep advertising them — + // without this the range list grows unboundedly on long + // connections. state.ackTracker.purgeBelow(frame.largestAcknowledged - frame.firstAckRange) + // Step 3 of `quic/plans/2026-05-04-control-frame-retransmit.md`: + // remove SentPacket entries for the PNs the peer acknowledged. + // Returned list is intentionally discarded for now — + // step 5 will use it to feed loss detection / RTT + // estimation; for step 3 we just drain. + drainAckedSentPackets(state.sentPackets, frame) } is CryptoFrame -> { diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPackets.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPackets.kt new file mode 100644 index 000000000..ec98d7b41 --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPackets.kt @@ -0,0 +1,84 @@ +/* + * 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.connection.recovery + +import com.vitorpamplona.quic.frame.AckFrame + +/** + * Walks the ACK ranges of an [AckFrame] in RFC 9000 §19.3.1 form and + * yields each ACK'd packet number, descending. The first range covers + * `[largestAcked - firstAckRange, largestAcked]`; each additional + * range's largest PN is `previousSmallest - gap - 2` and its smallest + * is `largest - ackRangeLength`. + * + * Ranges below packet number 0 are clamped — the spec disallows + * negative PNs, but defensive clamping keeps a malformed peer ACK + * from blowing up the iterator. + * + * The function does not allocate per-PN; it iterates by primitive + * `Long` and invokes [block] for each ACK'd PN. Hot path on every + * inbound ACK frame, so allocation matters. + */ +inline fun forEachAckedPacketNumber( + ack: AckFrame, + block: (Long) -> Unit, +) { + var currentLargest = ack.largestAcknowledged + var currentSmallest = currentLargest - ack.firstAckRange + if (currentSmallest < 0L) currentSmallest = 0L + var pn = currentLargest + while (pn >= currentSmallest) { + block(pn) + pn -= 1L + } + for (range in ack.additionalRanges) { + // Next range's largest = previousSmallest - gap - 2. + // Next range's smallest = largest - ackRangeLength. + val nextLargest = currentSmallest - range.gap - 2L + if (nextLargest < 0L) break + val nextSmallest = (nextLargest - range.ackRangeLength).coerceAtLeast(0L) + var pn2 = nextLargest + while (pn2 >= nextSmallest) { + block(pn2) + pn2 -= 1L + } + currentSmallest = nextSmallest + } +} + +/** + * Drain [sentPackets] of every entry whose packet number is covered + * by [ack]. Returns the drained packets so callers can pass their + * tokens to congestion-control / RTT estimation layers (step 5+ of + * `quic/plans/2026-05-04-control-frame-retransmit.md`). + * + * Caller must hold the connection lock. + */ +fun drainAckedSentPackets( + sentPackets: MutableMap, + ack: AckFrame, +): List { + val drained = mutableListOf() + forEachAckedPacketNumber(ack) { pn -> + sentPackets.remove(pn)?.let { drained += it } + } + return drained +} diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPacketsTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPacketsTest.kt new file mode 100644 index 000000000..7e86b0414 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPacketsTest.kt @@ -0,0 +1,238 @@ +/* + * 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.connection.recovery + +import com.vitorpamplona.quic.frame.AckFrame +import com.vitorpamplona.quic.frame.AckRange +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Step 3 of `quic/plans/2026-05-04-control-frame-retransmit.md`: + * `drainAckedSentPackets` removes SentPacket entries for every packet + * number covered by an inbound `AckFrame`'s ranges (RFC 9000 + * §19.3.1). Equivalent in scope to the `remove_acked` test in + * neqo's `recovery/mod.rs:1262`, plus a few range-walk edge cases. + */ +class AckedPacketsTest { + private fun sentPacket(pn: Long): SentPacket = + SentPacket( + packetNumber = pn, + sentAtMillis = 0L, + ackEliciting = true, + sizeBytes = 64, + tokens = listOf(RecoveryToken.MaxStreamsUni(maxStreams = pn * 10)), + ) + + @Test + fun simpleRange_drainsAllAckedPns() { + val sent = mutableMapOf() + for (pn in 0L..9L) sent[pn] = sentPacket(pn) + + // ACK [3..7]: largest=7, firstRange=4 → covers 3..7. + val ack = + AckFrame( + largestAcknowledged = 7L, + ackDelay = 0L, + firstAckRange = 4L, + ) + + val drained = drainAckedSentPackets(sent, ack) + assertEquals(5, drained.size, "5 PNs in [3..7]") + assertEquals(setOf(3L, 4L, 5L, 6L, 7L), drained.map { it.packetNumber }.toSet()) + // Map keeps everything outside the range. + assertEquals(setOf(0L, 1L, 2L, 8L, 9L), sent.keys) + } + + @Test + fun multipleRanges_drainsAllAckedPns() { + val sent = mutableMapOf() + for (pn in 0L..20L) sent[pn] = sentPacket(pn) + + // ACK ranges: [18..20] then [10..12] then [0..3]. + // Encoding (descending): + // firstRange covers [18..20] → largest=20, firstRange=2 + // gap before next: previousSmallest=18, nextLargest=12 ⇒ gap=18-12-2=4 + // nextLength: 12..10 ⇒ length=2 + // gap before next: previousSmallest=10, nextLargest=3 ⇒ gap=10-3-2=5 + // nextLength: 3..0 ⇒ length=3 + val ack = + AckFrame( + largestAcknowledged = 20L, + ackDelay = 0L, + firstAckRange = 2L, + additionalRanges = + listOf( + AckRange(gap = 4L, ackRangeLength = 2L), + AckRange(gap = 5L, ackRangeLength = 3L), + ), + ) + + val drained = drainAckedSentPackets(sent, ack) + val drainedPns = drained.map { it.packetNumber }.toSet() + assertEquals(setOf(18L, 19L, 20L, 10L, 11L, 12L, 0L, 1L, 2L, 3L), drainedPns) + // Map keeps everything outside the ranges. + assertEquals(setOf(4L, 5L, 6L, 7L, 8L, 9L, 13L, 14L, 15L, 16L, 17L), sent.keys) + } + + @Test + fun ackForUnsentPn_isNoOp() { + val sent = mutableMapOf() + sent[5L] = sentPacket(5L) + + // ACK PN 100 (never sent). drainAckedSentPackets should walk the + // range, find no matching key, and leave the map untouched. + // RFC 9000 §13.1 says ACK for unsent PN is a connection error — + // detecting that is the connection-state machine's job; this + // helper just drains, doesn't validate. + val ack = + AckFrame( + largestAcknowledged = 100L, + ackDelay = 0L, + firstAckRange = 0L, + ) + + val drained = drainAckedSentPackets(sent, ack) + assertEquals(0, drained.size) + assertEquals(1, sent.size) + assertNotNull(sent[5L]) + } + + @Test + fun emptyMap_returnsEmptyDrain() { + val sent = mutableMapOf() + val ack = + AckFrame( + largestAcknowledged = 5L, + ackDelay = 0L, + firstAckRange = 5L, + ) + val drained = drainAckedSentPackets(sent, ack) + assertTrue(drained.isEmpty()) + } + + @Test + fun singlePacketAck_drainsOne() { + val sent = mutableMapOf() + sent[42L] = sentPacket(42L) + + val ack = + AckFrame( + largestAcknowledged = 42L, + ackDelay = 0L, + firstAckRange = 0L, + ) + + val drained = drainAckedSentPackets(sent, ack) + assertEquals(1, drained.size) + assertEquals(42L, drained.single().packetNumber) + assertNull(sent[42L]) + } + + @Test + fun ackBoundary_pn0Inclusive() { + // First range [0..2] — must include PN 0 and not under-flow. + val sent = mutableMapOf() + for (pn in 0L..2L) sent[pn] = sentPacket(pn) + + val ack = + AckFrame( + largestAcknowledged = 2L, + ackDelay = 0L, + firstAckRange = 2L, + ) + + val drained = drainAckedSentPackets(sent, ack) + assertEquals(setOf(0L, 1L, 2L), drained.map { it.packetNumber }.toSet()) + assertTrue(sent.isEmpty()) + } + + @Test + fun forEachAckedPacketNumber_iteratesDescending() { + val ack = + AckFrame( + largestAcknowledged = 5L, + ackDelay = 0L, + firstAckRange = 2L, + ) + val pns = mutableListOf() + forEachAckedPacketNumber(ack) { pns += it } + // First range is [3..5]; iteration order is descending. + assertEquals(listOf(5L, 4L, 3L), pns) + } + + @Test + fun forEachAckedPacketNumber_acrossMultipleRanges_descending() { + // ACK [18..20] then [10..12] then [0..3] (same as multipleRanges_). + val ack = + AckFrame( + largestAcknowledged = 20L, + ackDelay = 0L, + firstAckRange = 2L, + additionalRanges = + listOf( + AckRange(gap = 4L, ackRangeLength = 2L), + AckRange(gap = 5L, ackRangeLength = 3L), + ), + ) + val pns = mutableListOf() + forEachAckedPacketNumber(ack) { pns += it } + assertEquals( + listOf(20L, 19L, 18L, 12L, 11L, 10L, 3L, 2L, 1L, 0L), + pns, + ) + } + + @Test + fun returnedDrain_preservesTokens() { + val sent = mutableMapOf() + sent[7L] = + SentPacket( + packetNumber = 7L, + sentAtMillis = 1_000L, + ackEliciting = true, + sizeBytes = 100, + tokens = + listOf( + RecoveryToken.MaxStreamsUni(maxStreams = 200L), + RecoveryToken.MaxData(maxData = 1_000_000L), + ), + ) + + val ack = + AckFrame( + largestAcknowledged = 7L, + ackDelay = 0L, + firstAckRange = 0L, + ) + val drained = drainAckedSentPackets(sent, ack) + + assertEquals(1, drained.size) + val sp = drained.single() + assertEquals(7L, sp.packetNumber) + assertEquals(2, sp.tokens.size) + assertEquals(RecoveryToken.MaxStreamsUni(200L), sp.tokens[0]) + assertEquals(RecoveryToken.MaxData(1_000_000L), sp.tokens[1]) + } +}