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 ccde73061..bc3dadef7 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -215,6 +215,24 @@ class QuicConnection( com.vitorpamplona.quic.connection.recovery .QuicLossDetection() + /** + * RFC 9002 §6.2 Probe Timeout signalling. When the driver loop's + * PTO timer fires (no ack-eliciting packet has been ACK'd in + * the PTO window), it sets [pendingPing] = true so the writer + * emits a PING frame on the next drain. The PING elicits an + * ACK from the peer; that ACK runs through loss detection and + * declares any in-flight packets lost, triggering retransmit. + */ + internal var pendingPing: Boolean = false + + /** + * RFC 9002 §6.2.2 consecutive PTO count. Incremented on each + * PTO expiration without an intervening ACK; reset to 0 when an + * inbound ACK acknowledges any ack-eliciting packet. The driver + * doubles its sleep between probes by `1 shl consecutivePtoCount`. + */ + internal var consecutivePtoCount: Int = 0 + /** * 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/QuicConnectionConfig.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt index 794021d36..f2cd6ba7b 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt @@ -37,39 +37,29 @@ data class QuicConnectionConfig( val initialMaxStreamDataUni: Long = 1L * 1024 * 1024, val initialMaxStreamsBidi: Long = 100L, /** - * Initial peer-initiated unidirectional stream limit. Sized to never - * trip the rolling [QuicConnectionWriter.appendFlowControlUpdates] - * extension path during a realistic audio-rooms broadcast — see - * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. + * Initial peer-initiated unidirectional stream limit. moq-rs's + * Quinn stack advertises `max_concurrent_uni_streams = 10000` + * for the same audio-rooms workload — every Opus group is a + * fresh peer-initiated uni stream — so we match. * - * Why a fixed-and-large initial value instead of relying on rolling - * extension: production tracing against `moq.nostrnests.com` showed - * that emitting `MAX_STREAMS_UNI` mid-connection silently breaks the - * relay's send path. The listener receives one more uni stream after - * the bump, then UDP goes dead at the kernel level - * (`udpRecvDatagrams` frozen) while our QUIC state still believes - * the connection is alive. The relay propagates the listener's - * disconnect to the publisher (`inbound SUBSCRIBE FIN'd` on the - * publisher side), but our stack never observes it — split-brain. - * Reproducible across runs. We don't yet know whether our frame is - * malformed, mis-sequenced relative to other frames in the packet, - * or hitting a moq-rs / Quinn bug — but we do know that not emitting - * the extension keeps the connection healthy. + * Earlier the value was set to 1 000 000 as a workaround for a + * production cliff against moq.nostrnests.com: emitting + * `MAX_STREAMS_UNI` mid-connection caused the relay to silently + * stop forwarding uni streams (see + * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`). + * That workaround dodged the bug at the cost of unbounded + * lifetime stream-id allocation per connection. * - * Capacity math, with [NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP] - * = 5 and Opus 20 ms: each group is one peer-initiated uni stream, - * so the relay opens ~10 streams/sec. The half-window threshold - * (`count + initialMaxStreamsUni/2 >= advertisedMaxStreamsUni`) - * trips at count = 500 000 streams, i.e. ~13.9 hours of continuous - * audio. Well past any realistic Nest duration. - * - * Memory cost: the [QuicConnection.streams] map currently grows for - * the connection's lifetime. At 10 streams/sec a 2-hour Nest leaves - * ~72k entries; per-stream overhead is small but unbounded growth - * over many hours is a known follow-up. For now this is a tolerable - * trade in exchange for not tripping the relay-side bug. + * The true fix landed in + * `quic/plans/2026-05-04-control-frame-retransmit.md`: + * RFC 9002 §6 loss detection + retransmit. With control-frame + * retransmit working, a single dropped MAX_STREAMS_UNI is + * recovered automatically — no need for the high-cap + * workaround. Lowered back to a moq-rs-matching default so the + * rolling-extension path (and its retransmit) actually + * exercises in production, validating the new code path. */ - val initialMaxStreamsUni: Long = 1_000_000L, + val initialMaxStreamsUni: Long = 10_000L, val maxIdleTimeoutMillis: Long = 30_000L, val maxUdpPayloadSize: Long = 1452L, val activeConnectionIdLimit: Long = 4L, diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt index 9fd62b324..48f198607 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -115,12 +115,13 @@ class QuicConnectionDriver( } private suspend fun sendLoop() { - // PTO budget: how long the loop will sleep before waking itself to - // check for retransmission opportunities. RFC 9002 §6.2 — initial - // PTO is roughly 3 × (smoothed RTT + max_ack_delay). We don't track - // RTT yet, so use a conservative fixed value that doubles on each - // consecutive timeout (Exponential backoff caps after ~6 timeouts). - var ptoMillis = 1_000L + // RFC 9002 §6.2 Probe Timeout. Once handshake is complete and + // we have an RTT estimate, the PTO duration is + // `smoothed_rtt + max(4*rttvar, 1ms) + max_ack_delay`, + // doubled by `1 shl consecutivePtoCount` per §6.2.2. Before + // the first RTT sample we fall back to a 1 s conservative + // floor (the same prior-shipping behavior, kept for + // handshake-timeout safety on lossy paths). while (connection.status != QuicConnection.Status.CLOSED) { connection.lock.withLock { while (true) { @@ -128,22 +129,32 @@ class QuicConnectionDriver( socket.send(out) } } + val ptoBaseMs = + if (connection.lossDetection.hasFirstRttSample) { + val maxAckDelayMs = connection.peerTransportParameters?.maxAckDelay ?: 0L + connection.lossDetection.ptoBaseMs(maxAckDelayMs).coerceAtLeast(1L) + } else { + 1_000L + } + val backoff = (1L shl connection.consecutivePtoCount.coerceAtMost(6)) + val ptoMillis = (ptoBaseMs * backoff).coerceAtMost(60_000L) // Suspend until either: a wakeup arrives, or the PTO timer expires. - // The PTO wake ensures a single lost ClientHello doesn't wedge - // the connection forever — eventually the loop wakes, the writer - // re-emits Initial CRYPTO that's still in the send buffer (since - // we don't free it until ACK), and the handshake retries. val woke = withTimeoutOrNull(ptoMillis) { sendWakeup.receive() Unit } - ptoMillis = - if (woke == null) { - (ptoMillis * 2).coerceAtMost(60_000L) - } else { - 1_000L + if (woke == null) { + // PTO fired. Set pendingPing so the writer emits a + // PING on the next drain (RFC 9002 §6.2.4 probe + // packet). The peer's ACK feeds loss detection + + // retransmit (steps 5–6). + connection.lock.withLock { + connection.pendingPing = true + connection.consecutivePtoCount = + (connection.consecutivePtoCount + 1).coerceAtMost(6) } + } } } 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 450c3c896..cefbfbc98 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -194,6 +194,10 @@ private fun dispatchFrames( ackDelayMs = ackDelayMs, nowMs = nowMillis, ) + // Step 7: reset PTO count on any new ack-eliciting + // ACK (RFC 9002 §6.2.1). The peer responded, so the + // exponential backoff resets. + conn.consecutivePtoCount = 0 } state.largestAckedPn?.let { largestAckedPn -> val lost = 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 0658307bb..23017cf88 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -31,6 +31,7 @@ import com.vitorpamplona.quic.frame.Frame import com.vitorpamplona.quic.frame.MaxDataFrame import com.vitorpamplona.quic.frame.MaxStreamDataFrame import com.vitorpamplona.quic.frame.MaxStreamsFrame +import com.vitorpamplona.quic.frame.PingFrame import com.vitorpamplona.quic.frame.StreamFrame import com.vitorpamplona.quic.frame.encodeFrames import com.vitorpamplona.quic.packet.LongHeaderPacket @@ -262,6 +263,18 @@ private fun buildApplicationPacket( tokens += RecoveryToken.Ack } + // Step 7: PTO probe. The driver sets `pendingPing` when its + // PTO timer fires without intervening ACKs. A PING is the + // smallest ack-eliciting frame (RFC 9000 §19.2) — its only + // purpose is to provoke an ACK from the peer, which then runs + // through loss detection (steps 5–6) and surfaces any + // outstanding losses for retransmit. The PING itself is not + // retransmitted on loss (RFC 9002 §A.9 PROBE_TIMEOUT skipped). + if (conn.pendingPing) { + frames += PingFrame + conn.pendingPing = false + } + // Re-credit the peer's send window when our receive offset has advanced // beyond half the previously-advertised limit. Emits MAX_STREAM_DATA per // stream and MAX_DATA at the connection level. 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 index c1eee5166..d07643eca 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/QuicLossDetection.kt @@ -160,6 +160,25 @@ class QuicLossDetection { return lost } + /** + * RFC 9002 §6.2.1 Probe Timeout duration: + * + * PTO = smoothed_rtt + max(4 * rttvar, kGranularity) + max_ack_delay + * + * For Application packets only; Initial / Handshake spaces use + * `max_ack_delay = 0` per §6.2.1. The connection passes the + * peer's [maxAckDelayMs] (from its transport parameters) — pass + * 0 if the peer hasn't advertised it yet. + * + * The result is doubled by the caller for each consecutive PTO + * expiration (§6.2.2 exponential backoff), capped so a long- + * silent connection doesn't wait forever between probes. + */ + fun ptoBaseMs(maxAckDelayMs: Long): Long { + val variancePart = (4L * rttVarMs).coerceAtLeast(GRANULARITY_MS) + return smoothedRttMs + variancePart + maxAckDelayMs + } + companion object { /** RFC 9002 §6.2.2 default initial RTT before a sample arrives. */ const val INITIAL_RTT_MS: Long = 333L diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/RetransmitIntegrationTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/RetransmitIntegrationTest.kt new file mode 100644 index 000000000..3eb1cf884 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/RetransmitIntegrationTest.kt @@ -0,0 +1,224 @@ +/* + * 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 + +import com.vitorpamplona.quic.connection.recovery.RecoveryToken +import com.vitorpamplona.quic.stream.StreamId +import com.vitorpamplona.quic.tls.InProcessTlsServer +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Step 8 of `quic/plans/2026-05-04-control-frame-retransmit.md`: + * end-to-end exercise of the retransmit pipeline. Drives the full + * chain (writer → SentPacket → loss detection → dispatch → drain) + * by simulating packet loss directly on QuicConnection state, since + * the in-process pipe doesn't model loss. + * + * This is the test that, before this work landed, could not have + * been written: there was no way to express "MAX_STREAMS_UNI was + * lost; verify it gets re-emitted". Now there is. + */ +class RetransmitIntegrationTest { + @Test + fun maxStreamsUni_lostByPacketThreshold_isRetransmitted() = + runBlocking { + val client = handshakedClient() + + // 1. Cross peer-uni half-window so the writer emits a + // MAX_STREAMS_UNI in its next drain. + crossPeerUniHalfWindow(client) + val capBefore = client.advertisedMaxStreamsUni + runCatching { drainOutbound(client, nowMillis = 1L) } + val capAfterFirstDrain = client.advertisedMaxStreamsUni + assertTrue(capAfterFirstDrain > capBefore, "writer must have advertised a higher cap") + + // 2. Locate the SentPacket carrying the MaxStreamsUni token. + val msuEntry = + client.application.sentPackets.entries.firstOrNull { entry -> + entry.value.tokens.any { it == RecoveryToken.MaxStreamsUni(maxStreams = capAfterFirstDrain) } + } + assertNotNull(msuEntry, "expected SentPacket with MaxStreamsUni token after first drain") + val msuPn = msuEntry.key + + // 3. Simulate ACK that newly-acks PN = msuPn + 4 (above the + // packet-threshold of 3). The MAX_STREAMS_UNI packet is + // ACK'd by reordering — its PN < largestAckedPn - + // PACKET_THRESHOLD ⇒ declared lost. + val futurePn = msuPn + 4L + client.lock.lock() + try { + // Inject a phantom SentPacket at futurePn so the loss + // detector has a credible "newly acked" reference, then + // ACK exactly that PN. + client.application.sentPackets[futurePn] = + com.vitorpamplona.quic.connection.recovery.SentPacket( + packetNumber = futurePn, + sentAtMillis = 1L, + ackEliciting = true, + sizeBytes = 64, + tokens = listOf(RecoveryToken.Ack), + ) + // Drain the ACK'd packet from the map (simulating the + // parser path). + client.application.sentPackets.remove(futurePn) + client.application.largestAckedPn = futurePn + client.application.largestAckedSentTimeMs = 1L + // 4. Run loss detection — msuPn is < futurePn - 3 ⇒ lost. + val lost = + client.lossDetection.detectAndRemoveLost( + sentPackets = client.application.sentPackets, + largestAckedPn = futurePn, + nowMs = 2L, + ) + val lostMsuPacket = lost.firstOrNull { it.packetNumber == msuPn } + assertNotNull(lostMsuPacket, "msuPn=$msuPn must be declared lost (largestAckedPn=$futurePn, threshold=3)") + // 5. Dispatch lost tokens — pendingMaxStreamsUni gets set. + client.onTokensLost(lostMsuPacket.tokens) + } finally { + client.lock.unlock() + } + assertEquals( + capAfterFirstDrain, + client.pendingMaxStreamsUni, + "lost MaxStreamsUni token must populate pendingMaxStreamsUni", + ) + + // 6. Drain again — writer must re-emit the MAX_STREAMS_UNI + // in a NEW packet (different PN). The pending* field + // is cleared after drain. + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 3L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + val retransmitEntry = + newEntries.firstOrNull { entry -> + entry.value.tokens.any { it == RecoveryToken.MaxStreamsUni(maxStreams = capAfterFirstDrain) } + } + assertNotNull( + retransmitEntry, + "retransmit must produce a fresh SentPacket carrying MaxStreamsUni; saw " + + newEntries.map { it.value.tokens.map { t -> t::class.simpleName } }, + ) + assertNotEquals(msuPn, retransmitEntry.key, "retransmit must use a NEW packet number") + assertNull(client.pendingMaxStreamsUni, "pendingMaxStreamsUni must be cleared after retransmit drain") + } + + @Test + fun lossDispatch_handlesSupersedeAcrossMultipleEmits() = + runBlocking { + // Two consecutive MAX_STREAMS_UNI emissions; the OLDER one + // is declared lost. Because the newer extension already + // covers it, the supersede check (in onTokensLost) drops + // the older lost token without populating pending*. + val client = handshakedClient() + + // First bump. + crossPeerUniHalfWindow(client) + runCatching { drainOutbound(client, nowMillis = 1L) } + val firstCap = client.advertisedMaxStreamsUni + + // Second bump: open more peer-uni streams to cross the + // (already extended) threshold again. + client.lock.lock() + try { + client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 2)) + client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 3)) + client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 4)) + } finally { + client.lock.unlock() + } + runCatching { drainOutbound(client, nowMillis = 2L) } + val secondCap = client.advertisedMaxStreamsUni + assertTrue(secondCap > firstCap, "second drain must advertise a still-higher cap; saw $firstCap → $secondCap") + + // Now declare the FIRST emit lost via direct dispatch. + client.lock.lock() + try { + client.onTokensLost(listOf(RecoveryToken.MaxStreamsUni(maxStreams = firstCap))) + } finally { + client.lock.unlock() + } + // Supersede check: firstCap != advertisedMaxStreamsUni (now == secondCap), + // so pending must remain null. + assertNull(client.pendingMaxStreamsUni, "older lost extension is superseded by newer emit; no retransmit") + } + + private fun handshakedClient(): QuicConnection = + runBlocking { + val client = + QuicConnection( + serverName = "example.test", + config = + QuicConnectionConfig( + initialMaxStreamsUni = 4, + initialMaxStreamsBidi = 4, + ), + tlsCertificateValidator = + com.vitorpamplona.quic.tls + .PermissiveCertificateValidator(), + ) + val serverScid = ConnectionId.random(8) + val tlsServer = + InProcessTlsServer( + transportParameters = + TransportParameters( + initialMaxData = 1_000_000, + initialMaxStreamDataBidiLocal = 100_000, + initialMaxStreamDataBidiRemote = 100_000, + initialMaxStreamDataUni = 100_000, + initialMaxStreamsBidi = 100, + initialMaxStreamsUni = 100, + initialSourceConnectionId = serverScid.bytes, + originalDestinationConnectionId = client.destinationConnectionId.bytes, + ).encode(), + ) + val pipe = + InMemoryQuicPipe( + client = client, + initialDcid = client.destinationConnectionId.bytes, + serverScid = serverScid, + tlsServer = tlsServer, + ) + client.start() + pipe.drive(maxRounds = 16) + assertEquals(QuicConnection.Status.CONNECTED, client.status) + client + } + + private fun crossPeerUniHalfWindow(client: QuicConnection) = + runBlocking { + client.lock.lock() + try { + client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 0)) + client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 1)) + } finally { + client.lock.unlock() + } + } +} diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/PtoTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/PtoTest.kt new file mode 100644 index 000000000..0290f97e8 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/PtoTest.kt @@ -0,0 +1,82 @@ +/* + * 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.assertTrue + +/** + * Step 7 of `quic/plans/2026-05-04-control-frame-retransmit.md`: + * RFC 9002 §6.2 Probe Timeout duration calculation. Mirrors neqo's + * PTO tests (`pto_works_basic`, `pto_state_count`, etc.) at the + * algorithm level — the driver-loop integration is exercised by + * the existing connection tests. + */ +class PtoTest { + @Test + fun ptoBeforeFirstRttSample_usesInitialDefault() { + val ld = QuicLossDetection() + // Before any sample: smoothed_rtt = 333, rttvar = 333/2 = 166. + // PTO = 333 + max(4*166, 1) + 0 = 333 + 664 = 997. + assertEquals(997L, ld.ptoBaseMs(maxAckDelayMs = 0L)) + } + + @Test + fun ptoIncludesMaxAckDelay() { + val ld = QuicLossDetection() + assertEquals(997L + 25L, ld.ptoBaseMs(maxAckDelayMs = 25L)) + } + + @Test + fun ptoAfterRttSample_usesNewSmoothedRtt() { + val ld = QuicLossDetection() + ld.onRttSample(largestAckedSentTimeMs = 0L, ackDelayMs = 0L, nowMs = 100L) + // After first sample: smoothed = 100, rttvar = 50. + // PTO = 100 + max(4*50, 1) + 0 = 100 + 200 = 300. + assertEquals(300L, ld.ptoBaseMs(maxAckDelayMs = 0L)) + } + + @Test + fun ptoFloors4RttVarAtGranularity() { + val ld = QuicLossDetection() + // Two identical samples drive rttvar toward 0. Then PTO should + // floor `4 * rttvar` at 1 ms granularity. + ld.onRttSample(largestAckedSentTimeMs = 0L, ackDelayMs = 0L, nowMs = 100L) + // rttvar after first sample = 50. + ld.onRttSample(largestAckedSentTimeMs = 100L, ackDelayMs = 0L, nowMs = 200L) + // smoothed = (7*100 + 100)/8 = 100. rttvar = (3*50 + 0)/4 = 37. + // 4*37 = 148, > 1ms. OK floor not exercised yet. + // Many identical samples drive rttvar → 0: + var t = 200L + repeat(40) { + ld.onRttSample(largestAckedSentTimeMs = t, ackDelayMs = 0L, nowMs = t + 100L) + t += 100L + } + // After many samples: rttvar should be tiny but positive. PTO + // contribution from 4*rttvar might fall below 1; check the + // floor by comparing PTO to smoothed_rtt + max_ack_delay only + // (i.e. the variance contribution is at least 1). + val smoothed = ld.smoothedRttMs + val pto = ld.ptoBaseMs(maxAckDelayMs = 0L) + assertTrue(pto >= smoothed + 1L, "PTO must include at least 1ms of variance: pto=$pto smoothed=$smoothed") + } +}