diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt index c123199ff..919c049a1 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt @@ -50,4 +50,20 @@ class LevelState { * [QuicConnection.lock]. */ val sentPackets: MutableMap = HashMap() + + /** + * RFC 9002 §6.1 largest acknowledged packet number observed + * by an inbound ACK in this space. Updated only when the ACK + * advances it — duplicate ACKs do not move the value. + * Used as the high-water mark for packet-threshold loss detection. + */ + var largestAckedPn: Long? = null + + /** + * Send time (epoch ms, from the writer's `nowMillis` source) of + * the SentPacket whose PN is [largestAckedPn]. Used to compute + * the RTT sample on a newly-advancing ACK (RFC 9002 §5.2). + * Null until an ACK arrives. + */ + var largestAckedSentTimeMs: Long? = null } 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 d19c0c9fc..1754bf922 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -205,6 +205,16 @@ class QuicConnection( */ internal val pendingMaxStreamData: MutableMap = HashMap() + /** + * RFC 9002 RTT estimator + loss-detection algorithm. Single + * shared instance per connection (RTT is per-path; we model a + * single path). Per-space `largestAcked*` lives on + * [LevelState]. Step 6 wires the loss-detection callback. + */ + internal val lossDetection: com.vitorpamplona.quic.connection.recovery.QuicLossDetection = + com.vitorpamplona.quic.connection.recovery + .QuicLossDetection() + /** * Optional supplier of underlying UDP-socket counters. Wired by the * platform-specific driver since `UdpSocket`'s counters are 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 d81cee1bb..1b797c3eb 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -170,12 +170,43 @@ private fun dispatchFrames( // 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) + // Step 5 of `quic/plans/2026-05-04-control-frame-retransmit.md`: + // (a) snapshot the send time of the largest-acked PN + // BEFORE drain so we can update RTT, (b) drain ACK'd + // packets, (c) if the ACK advanced largestAckedPn AND + // any drained packet was ack-eliciting, update RTT, + // (d) detect-and-remove lost packets. The lost-token + // dispatch (step 6) is a TODO — for step 5 we drop the + // returned list on the floor. + val largestSentTime = state.sentPackets[frame.largestAcknowledged]?.sentAtMillis + val drained = drainAckedSentPackets(state.sentPackets, frame) + val advancedLargest = + state.largestAckedPn?.let { it < frame.largestAcknowledged } ?: true + if (advancedLargest) { + state.largestAckedPn = frame.largestAcknowledged + state.largestAckedSentTimeMs = largestSentTime + } + if (advancedLargest && largestSentTime != null && drained.any { it.ackEliciting }) { + val ackDelayUs = frame.ackDelay shl conn.config.ackDelayExponent.toInt() + val ackDelayMs = ackDelayUs / 1_000L + conn.lossDetection.onRttSample( + largestAckedSentTimeMs = largestSentTime, + ackDelayMs = ackDelayMs, + nowMs = nowMillis, + ) + } + state.largestAckedPn?.let { largestAckedPn -> + val lost = + conn.lossDetection.detectAndRemoveLost( + sentPackets = state.sentPackets, + largestAckedPn = largestAckedPn, + nowMs = nowMillis, + ) + + // Step 6 will dispatch tokens here. Drop for now. + @Suppress("UNUSED_VARIABLE") + val unusedForStep6 = lost + } } is CryptoFrame -> { diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetection.kt new file mode 100644 index 000000000..c1eee5166 --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetection.kt @@ -0,0 +1,181 @@ +/* + * 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 kotlin.math.abs + +/** + * RFC 9002 §5–§6 loss detection — RTT estimation + packet/time + * threshold loss declaration. Mirrors the algorithm Firefox neqo + * implements at `neqo-transport/src/recovery/mod.rs` and `rtt.rs`. + * + * State is per-connection (RTT estimates are per-path; we model a + * single path). Per-encryption-space state — `largestAckedPn`, + * `largestAckedSentTime` — lives on [com.vitorpamplona.quic.connection.LevelState]. + * + * Step 5 of `quic/plans/2026-05-04-control-frame-retransmit.md`: this + * file implements the algorithm; the parser invokes it on every + * inbound ACK; step 6 dispatches the returned lost-token list to + * the `pending*` setters. + */ +class QuicLossDetection { + /** + * RFC 9002 §5.3. Smoothed RTT estimate. Initialised to + * [INITIAL_RTT_MS] until the first sample arrives. Updated as + * `smoothed_rtt = (7/8) * smoothed_rtt + (1/8) * adjusted_rtt`. + */ + var smoothedRttMs: Long = INITIAL_RTT_MS + private set + + /** + * RFC 9002 §5.3. RTT variance estimate. Initialised to half the + * initial RTT. Updated as + * `rttvar = (3/4) * rttvar + (1/4) * |smoothed_rtt - adjusted_rtt|`. + */ + var rttVarMs: Long = INITIAL_RTT_MS / 2 + private set + + /** + * RFC 9002 §5.2. Latest RTT sample, before ack-delay adjustment. + * Tracks the most recent observation so callers can compute + * `max(latestRtt, smoothedRtt)` for the time-threshold loss check. + */ + var latestRttMs: Long = INITIAL_RTT_MS + private set + + /** + * RFC 9002 §5.2. Minimum RTT observed on this path. Used to + * clamp ack-delay so a peer can't artificially inflate our RTT + * estimate by reporting a large ack-delay. + */ + var minRttMs: Long = Long.MAX_VALUE + private set + + /** True after the first valid RTT sample has been processed. */ + var hasFirstRttSample: Boolean = false + private set + + /** + * Update RTT estimates from a new ACK that newly-acknowledged + * the largest packet number. Per RFC 9002 §5.2, an RTT sample + * is only generated when: + * - the largest acknowledged packet number is newly acknowledged, and + * - at least one of the newly acknowledged packets was ack-eliciting. + * + * The caller must enforce both conditions before calling this. + */ + fun onRttSample( + largestAckedSentTimeMs: Long, + ackDelayMs: Long, + nowMs: Long, + ) { + val rawSampleMs = nowMs - largestAckedSentTimeMs + if (rawSampleMs < 0L) return // clock skew; ignore + if (!hasFirstRttSample) { + minRttMs = rawSampleMs + smoothedRttMs = rawSampleMs + rttVarMs = rawSampleMs / 2 + latestRttMs = rawSampleMs + hasFirstRttSample = true + return + } + if (rawSampleMs < minRttMs) minRttMs = rawSampleMs + // Adjust by ack-delay, clamped so that adjusted_rtt >= min_rtt. + // Without the clamp, a peer reporting a fake high ack-delay can + // push smoothed_rtt below min_rtt (RFC 9002 §5.3). + val adjusted = + if (rawSampleMs - ackDelayMs >= minRttMs) { + rawSampleMs - ackDelayMs + } else { + rawSampleMs + } + latestRttMs = adjusted + rttVarMs = (3L * rttVarMs + abs(smoothedRttMs - adjusted)) / 4L + smoothedRttMs = (7L * smoothedRttMs + adjusted) / 8L + } + + /** + * RFC 9002 §6.1.2. Loss-detection time threshold: + * `loss_delay = max(latest_rtt, smoothed_rtt) * (9/8)` + * clamped to at least [GRANULARITY_MS]. + */ + fun lossDelayMs(): Long { + val maxRtt = if (latestRttMs > smoothedRttMs) latestRttMs else smoothedRttMs + val scaled = (maxRtt * TIME_THRESHOLD_NUM) / TIME_THRESHOLD_DEN + return if (scaled > GRANULARITY_MS) scaled else GRANULARITY_MS + } + + /** + * Detect packets lost in [sentPackets] and remove them. Called + * after the parser drains ACK'd packets — the surviving entries + * are the in-flight set, plus any older packets that haven't + * been ACK'd yet. RFC 9002 §6.1 declares a packet lost iff: + * + * - it has a smaller PN than [largestAckedPn] AND + * - it was sent at least [PACKET_THRESHOLD] PNs before + * [largestAckedPn], OR + * - it was sent more than [lossDelayMs] ago. + * + * Returns the list of lost packets in arbitrary order. Caller + * dispatches their [SentPacket.tokens] (step 6). + */ + fun detectAndRemoveLost( + sentPackets: MutableMap, + largestAckedPn: Long, + nowMs: Long, + ): List { + if (sentPackets.isEmpty()) return emptyList() + val lossDelay = lossDelayMs() + val lossThresholdSentMs = nowMs - lossDelay + val lost = mutableListOf() + val it = sentPackets.entries.iterator() + while (it.hasNext()) { + val (pn, pkt) = it.next() + if (pn >= largestAckedPn) continue + val packetThresholdLost = pn < largestAckedPn - PACKET_THRESHOLD + val timeThresholdLost = pkt.sentAtMillis <= lossThresholdSentMs + if (packetThresholdLost || timeThresholdLost) { + lost += pkt + it.remove() + } + } + return lost + } + + companion object { + /** RFC 9002 §6.2.2 default initial RTT before a sample arrives. */ + const val INITIAL_RTT_MS: Long = 333L + + /** RFC 9002 §6.1.1 packet-reordering threshold (number of PNs). */ + const val PACKET_THRESHOLD: Long = 3L + + /** + * RFC 9002 §6.1.2 time-threshold multiplier. The scaled max + * RTT is `max_rtt * NUM / DEN`; with NUM=9, DEN=8 the + * threshold is `max_rtt * 9/8 = max_rtt + max_rtt/8`. + */ + const val TIME_THRESHOLD_NUM: Long = 9L + const val TIME_THRESHOLD_DEN: Long = 8L + + /** RFC 9002 §6.1.2 granularity floor. Most stacks use 1 ms. */ + const val GRANULARITY_MS: Long = 1L + } +} diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetectionTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetectionTest.kt new file mode 100644 index 000000000..a5f8627be --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetectionTest.kt @@ -0,0 +1,197 @@ +/* + * 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 kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Step 5 of `quic/plans/2026-05-04-control-frame-retransmit.md`: + * RFC 9002 §5–§6 RTT + loss detection. Mirrors the subset of + * neqo's `recovery/mod.rs` tests that don't depend on PTO (step 7) + * or CRYPTO/handshake-space behavior (out of scope). + */ +class QuicLossDetectionTest { + private fun sentPacket( + pn: Long, + sentAt: Long, + ackEliciting: Boolean = true, + ): SentPacket = + SentPacket( + packetNumber = pn, + sentAtMillis = sentAt, + ackEliciting = ackEliciting, + sizeBytes = 64, + tokens = listOf(RecoveryToken.MaxStreamsUni(maxStreams = pn * 100L)), + ) + + @Test + fun firstRttSample_setsAllRttFieldsAtomically() { + val ld = QuicLossDetection() + ld.onRttSample(largestAckedSentTimeMs = 1_000L, ackDelayMs = 0L, nowMs = 1_050L) + assertTrue(ld.hasFirstRttSample) + assertEquals(50L, ld.smoothedRttMs, "first sample becomes smoothed_rtt") + assertEquals(50L, ld.minRttMs) + assertEquals(50L, ld.latestRttMs) + assertEquals(25L, ld.rttVarMs, "rttvar = sample/2 on first sample") + } + + @Test + fun secondRttSample_movesSmoothedRttTowardSample() { + val ld = QuicLossDetection() + ld.onRttSample(largestAckedSentTimeMs = 0L, ackDelayMs = 0L, nowMs = 100L) + // First sample: smoothed=100, rttvar=50, min=100. + // Second sample at sample=120 (raw): adjusted=120 (no ack-delay). + // smoothed' = (7*100 + 120)/8 = 820/8 = 102 (integer) + // rttvar' = (3*50 + |100-120|)/4 = (150 + 20)/4 = 42 + ld.onRttSample(largestAckedSentTimeMs = 100L, ackDelayMs = 0L, nowMs = 220L) + assertEquals(102L, ld.smoothedRttMs) + assertEquals(42L, ld.rttVarMs) + assertEquals(100L, ld.minRttMs, "minRtt only decreases") + } + + @Test + fun ackDelay_clampedAgainstMinRtt() { + val ld = QuicLossDetection() + // First sample establishes minRtt=50. + ld.onRttSample(largestAckedSentTimeMs = 0L, ackDelayMs = 0L, nowMs = 50L) + // Second sample raw=100, peer reports ackDelay=80. + // adjusted would be 100 - 80 = 20 < minRtt(50), so clamp: + // adjusted := raw = 100. + ld.onRttSample(largestAckedSentTimeMs = 100L, ackDelayMs = 80L, nowMs = 200L) + assertEquals(100L, ld.latestRttMs) + // smoothed' = (7*50 + 100)/8 = 56 + assertEquals(56L, ld.smoothedRttMs) + } + + @Test + fun negativeRttSample_isIgnored() { + // Clock skew or packet number reuse — sample is negative; ignore. + val ld = QuicLossDetection() + ld.onRttSample(largestAckedSentTimeMs = 100L, ackDelayMs = 0L, nowMs = 50L) + assertEquals(false, ld.hasFirstRttSample) + assertEquals(QuicLossDetection.INITIAL_RTT_MS, ld.smoothedRttMs) + } + + @Test + fun lossDelay_floor() { + val ld = QuicLossDetection() + // Initial: smoothed=333, latest=333. Loss delay = 333*9/8 = 374. + assertEquals(374L, ld.lossDelayMs()) + } + + @Test + fun packetThresholdLost_removesPacketsBelowThreshold() { + val ld = QuicLossDetection() + val sent = mutableMapOf() + for (pn in 0L..10L) sent[pn] = sentPacket(pn, sentAt = 0L) + // largestAckedPn=10. Packet threshold: lost iff pn < 10 - 3 = 7. + // Pns 0..6 ⇒ 7 packets lost. Pns 7..10 not lost (pn >= 7). + // (Pn 10 is the largest-acked itself — not in flight; it was just removed by drain. + // We model "after-drain" by removing 10 from sent before calling detectAndRemoveLost.) + sent.remove(10L) + val lost = ld.detectAndRemoveLost(sent, largestAckedPn = 10L, nowMs = 1L) + // sentAt=0, nowMs=1, lossDelayMs=374 → time threshold cutoff at -373: nothing lost by time threshold + // (sentAt 0 is NOT <= -373). But packet threshold removes pns 0..6. + assertEquals(setOf(0L, 1L, 2L, 3L, 4L, 5L, 6L), lost.map { it.packetNumber }.toSet()) + assertEquals(setOf(7L, 8L, 9L), sent.keys, "pns 7..9 remain in flight; pn 10 was drained earlier") + } + + @Test + fun timeThresholdLost_removesPacketsSentTooLongAgo() { + val ld = QuicLossDetection() + // Force an RTT sample so loss delay is small (10ms*9/8 = 11ms with floor 1). + ld.onRttSample(largestAckedSentTimeMs = 0L, ackDelayMs = 0L, nowMs = 10L) + assertEquals(11L, ld.lossDelayMs(), "10*9/8 = 11.25 → 11") + + val sent = mutableMapOf() + sent[0L] = sentPacket(0L, sentAt = 100L) // old: 100 + 11 = 111, now=200 ⇒ lost (sentAt<=189) + sent[1L] = sentPacket(1L, sentAt = 195L) // recent: 195 + 11 = 206, now=200 ⇒ not lost + // pn 2 is the largest-acked, drained earlier. + + val lost = ld.detectAndRemoveLost(sent, largestAckedPn = 2L, nowMs = 200L) + assertEquals(listOf(0L), lost.map { it.packetNumber }) + assertEquals(setOf(1L), sent.keys) + } + + @Test + fun packetEqualToLargestAcked_notLost() { + // Edge case: pn == largestAckedPn shouldn't be in the in-flight + // map at this point (drain already removed it), but defensively + // detectAndRemoveLost should not declare it lost either. + val ld = QuicLossDetection() + val sent = mutableMapOf() + sent[5L] = sentPacket(5L, sentAt = 0L) + val lost = ld.detectAndRemoveLost(sent, largestAckedPn = 5L, nowMs = 1L) + assertTrue(lost.isEmpty()) + assertNotNull(sent[5L]) + } + + @Test + fun emptyMap_returnsEmpty() { + val ld = QuicLossDetection() + val sent = mutableMapOf() + val lost = ld.detectAndRemoveLost(sent, largestAckedPn = 100L, nowMs = 1L) + assertTrue(lost.isEmpty()) + } + + @Test + fun lostPackets_carryOriginalTokens() { + val ld = QuicLossDetection() + val sent = mutableMapOf() + sent[0L] = + SentPacket( + packetNumber = 0L, + sentAtMillis = 0L, + ackEliciting = true, + sizeBytes = 100, + tokens = + listOf( + RecoveryToken.MaxStreamsUni(maxStreams = 150L), + RecoveryToken.MaxData(maxData = 5_000L), + ), + ) + // PN 0 is below largestAckedPn=10 - threshold(3) = 7, so lost by packet threshold. + val lost = ld.detectAndRemoveLost(sent, largestAckedPn = 10L, nowMs = 1L) + assertEquals(1, lost.size) + assertEquals(2, lost.single().tokens.size) + assertEquals(RecoveryToken.MaxStreamsUni(150L), lost.single().tokens[0]) + assertEquals(RecoveryToken.MaxData(5_000L), lost.single().tokens[1]) + assertNull(sent[0L]) + } + + @Test + fun packetThresholdAndTimeThresholdMatch_singleRemoval() { + // Verify packet is only dropped from the map once even when both + // thresholds say "lost" (defensive — Map.iterator.remove() should + // be called exactly once per entry). + val ld = QuicLossDetection() + ld.onRttSample(largestAckedSentTimeMs = 0L, ackDelayMs = 0L, nowMs = 10L) + val sent = mutableMapOf() + sent[0L] = sentPacket(0L, sentAt = 0L) // old AND below threshold + val lost = ld.detectAndRemoveLost(sent, largestAckedPn = 10L, nowMs = 200L) + assertEquals(1, lost.size) + assertTrue(sent.isEmpty()) + } +}