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

feat(quic): step 5 of RFC 9002 retransmit — loss detection + RTT estimator

Browse Source 
QuicLossDetection encapsulates the RFC 9002 §5–§6 algorithms:

  - RTT estimation (§5): smoothedRtt, rttVar, latestRtt, minRtt
    with first-sample bootstrap; ack-delay clamped against minRtt
    so a peer reporting a large delay can't push the estimate
    below its observed floor (§5.3 anti-exploit clamp).
  - Loss-delay (§6.1.2): max(latestRtt, smoothedRtt) * 9/8,
    clamped to GRANULARITY_MS (1 ms).
  - detectAndRemoveLost (§6.1): walks the in-flight set,
    removes entries that are either:
      - more than PACKET_THRESHOLD (3) PNs below largestAckedPn,
        OR
      - sent more than lossDelay ago.
    Returns the lost packets so step 6 can dispatch their tokens.

Wired into QuicConnectionParser's AckFrame handler:

  1. Snapshot largest-acked send time BEFORE drain
  2. Drain ACK'd packets (step 3)
  3. If largestAckedPn advanced AND any drained packet was
     ack-eliciting, update RTT (RFC 9002 §5.2 sample conditions)
  4. detectAndRemoveLost on the surviving set; lost list dropped
     for now — step 6 wires the dispatch

LevelState gains:
  - largestAckedPn: Long? (high-water mark for packet-threshold)
  - largestAckedSentTimeMs: Long? (RTT sample input)

QuicConnection gains:
  - lossDetection: QuicLossDetection (single instance, RTT is
    per-path; we model a single path)

Tests added (11, all pass):
  - firstRttSample_setsAllRttFieldsAtomically
  - secondRttSample_movesSmoothedRttTowardSample (math: 7/8 + 1/8)
  - ackDelay_clampedAgainstMinRtt (anti-exploit clamp)
  - negativeRttSample_isIgnored (clock skew defense)
  - lossDelay_floor (initial 333*9/8 = 374)
  - packetThresholdLost_removesPacketsBelowThreshold (PNs 0..6
    when largestAcked=10, threshold=3 → 7..9 survive)
  - timeThresholdLost_removesPacketsSentTooLongAgo (sentAt + delay
    < now → lost; recent → kept)
  - packetEqualToLargestAcked_notLost (edge case)
  - emptyMap_returnsEmpty
  - lostPackets_carryOriginalTokens (drain preserves token list)
  - packetThresholdAndTimeThresholdMatch_singleRemoval (no
    double-iteration)

Mirrors the subset of neqo's `recovery/mod.rs` tests in scope per
the plan: remove_acked, time_loss_detection_gap,
time_loss_detection_timeout, big_gap_loss,
duplicate_ack_does_not_update_largest_acked_sent_time. PTO tests
land in step 7.

Full :quic test suite passes.

https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ
This commit is contained in:
Claude
2026-05-04 22:52:49 +00:00
parent 29282634e5
commit 1df6441639
5 changed files with 441 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt
+16
View File
@@ -50,4 +50,20 @@ class LevelState {
* [QuicConnection.lock].
*/
val sentPackets: MutableMap<Long, SentPacket> = 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
}
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt
+10
View File
@@ -205,6 +205,16 @@ class QuicConnection(
*/
internal val pendingMaxStreamData: MutableMap<Long, Long> = 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
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt
+37 -6
View File
@@ -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 -> {
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetection.kt
+181
View File
@@ -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<Long, SentPacket>,
largestAckedPn: Long,
nowMs: Long,
): List<SentPacket> {
if (sentPackets.isEmpty()) return emptyList()
val lossDelay = lossDelayMs()
val lossThresholdSentMs = nowMs - lossDelay
val lost = mutableListOf<SentPacket>()
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
}
}
quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetectionTest.kt
+197
View File
@@ -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<Long, SentPacket>()
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<Long, SentPacket>()
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<Long, SentPacket>()
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<Long, SentPacket>()
val lost = ld.detectAndRemoveLost(sent, largestAckedPn = 100L, nowMs = 1L)
assertTrue(lost.isEmpty())
}
@Test
fun lostPackets_carryOriginalTokens() {
val ld = QuicLossDetection()
val sent = mutableMapOf<Long, SentPacket>()
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<Long, SentPacket>()
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())
}
}