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 041974e02..d19c0c9fc 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -172,6 +172,39 @@ class QuicConnection( /** Bidi counterpart of [advertisedMaxStreamsUni]. */ internal var advertisedMaxStreamsBidi: Long = config.initialMaxStreamsBidi + /** + * RFC 9002 retransmit pending-state for the receive-side flow- + * control extensions. Non-null means "the last MAX_STREAMS_UNI + * we sent was lost; the writer should re-emit on its next pass". + * + * The value held is the limit that was lost — meaningful only + * when it equals [advertisedMaxStreamsUni], which is the + * supersede-check from neqo's `fc.rs::frame_lost` (if a newer, + * higher extension has since gone out, the older lost frame is + * irrelevant). Step 6 of + * `quic/plans/2026-05-04-control-frame-retransmit.md` populates + * this field via the loss dispatcher; step 4 (this commit) wires + * the writer to drain it before running the rolling-extension + * threshold check. + * + * Caller of any read/write must hold [lock]. + */ + internal var pendingMaxStreamsUni: Long? = null + + /** Bidi counterpart of [pendingMaxStreamsUni]. */ + internal var pendingMaxStreamsBidi: Long? = null + + /** Connection-level MAX_DATA counterpart of [pendingMaxStreamsUni]. */ + internal var pendingMaxData: Long? = null + + /** + * Per-stream MAX_STREAM_DATA pending-state. Keyed by stream id. + * Same semantics as [pendingMaxStreamsUni]: present iff a + * MAX_STREAM_DATA for that stream was lost and hasn't been + * superseded by a higher emit. Wired by step 6. + */ + internal val pendingMaxStreamData: MutableMap = HashMap() + /** * 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/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 40ea62ddc..0658307bb 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -386,6 +386,42 @@ private fun appendFlowControlUpdates( frames: MutableList, tokens: MutableList, ) { + // Step 4: re-emit any retransmittable extension whose carrying + // packet was declared lost (the `pending*` fields are populated + // by step 6's loss dispatcher; for now they default to null). + // Each pending entry is consumed by emitting a fresh frame with + // the same value plus a fresh token, so the new packet's loss + // can be re-detected if the wire drops this one too. The + // supersede-check (`token.value == advertised*`) lives on the + // setter side in step 6 — by the time `pending*` is set, the + // value is known to still match the current advertised cap. + val pendingUni = conn.pendingMaxStreamsUni + if (pendingUni != null) { + frames += MaxStreamsFrame(bidi = false, maxStreams = pendingUni) + tokens += RecoveryToken.MaxStreamsUni(maxStreams = pendingUni) + conn.pendingMaxStreamsUni = null + } + val pendingBidi = conn.pendingMaxStreamsBidi + if (pendingBidi != null) { + frames += MaxStreamsFrame(bidi = true, maxStreams = pendingBidi) + tokens += RecoveryToken.MaxStreamsBidi(maxStreams = pendingBidi) + conn.pendingMaxStreamsBidi = null + } + val pendingData = conn.pendingMaxData + if (pendingData != null) { + frames += MaxDataFrame(pendingData) + tokens += RecoveryToken.MaxData(maxData = pendingData) + conn.pendingMaxData = null + } + if (conn.pendingMaxStreamData.isNotEmpty()) { + // Iterate over a snapshot so we can mutate the map safely. + val pendingStreamEntries = conn.pendingMaxStreamData.entries.toList() + for ((streamId, maxData) in pendingStreamEntries) { + frames += MaxStreamDataFrame(streamId, maxData) + tokens += RecoveryToken.MaxStreamData(streamId = streamId, maxData = maxData) + } + conn.pendingMaxStreamData.clear() + } val cfg = conn.config var totalRecvAdvanced = 0L // Round-4 perf #9 + round-5 #9: walk the streams via the index-friendly diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PendingFlowControlEmitTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PendingFlowControlEmitTest.kt new file mode 100644 index 000000000..1c42c9aeb --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PendingFlowControlEmitTest.kt @@ -0,0 +1,305 @@ +/* + * 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.tls.InProcessTlsServer +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Step 4 of `quic/plans/2026-05-04-control-frame-retransmit.md`: + * the writer drains `QuicConnection.pendingMax*` fields ahead of the + * normal threshold check, re-emitting a fresh frame + token for each + * pending entry and clearing it. + * + * Mirrors neqo's `fc.rs` retransmit tests + * (`need_max_allowed_frame_after_loss`, `lost_after_increase`, + * `multiple_retries_after_frame_pending_is_set`, + * `new_retired_before_loss`). The `pending*` fields are populated + * directly here; step 6 will populate them via the loss dispatcher. + */ +class PendingFlowControlEmitTest { + @Test + fun pendingMaxStreamsUni_drainEmitsFrameAndToken() = + runBlocking { + val client = handshakedClient() + client.lock.lock() + try { + client.pendingMaxStreamsUni = 150L + } finally { + client.lock.unlock() + } + + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + + // Some new SentPacket must carry a MaxStreamsUni token + // with the pending value. + val withRetransmit = + newEntries.firstOrNull { entry -> + entry.value.tokens.any { it == RecoveryToken.MaxStreamsUni(maxStreams = 150L) } + } + assertNotNull( + withRetransmit, + "expected re-emit of MaxStreamsUni(150) but saw " + + newEntries.map { it.value.tokens.map { t -> t::class.simpleName } }, + ) + // Pending must be cleared after emit. + assertNull(client.pendingMaxStreamsUni) + } + + @Test + fun pendingMaxStreamsBidi_drainEmitsFrameAndToken(): Unit = + runBlocking { + val client = handshakedClient() + client.lock.lock() + try { + client.pendingMaxStreamsBidi = 200L + } finally { + client.lock.unlock() + } + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + val withRetransmit = + newEntries.firstOrNull { entry -> + entry.value.tokens.any { it == RecoveryToken.MaxStreamsBidi(maxStreams = 200L) } + } + assertNotNull(withRetransmit, "expected re-emit of MaxStreamsBidi(200)") + assertNull(client.pendingMaxStreamsBidi) + } + + @Test + fun pendingMaxData_drainEmitsFrameAndToken(): Unit = + runBlocking { + val client = handshakedClient() + client.lock.lock() + try { + client.pendingMaxData = 5_000_000L + } finally { + client.lock.unlock() + } + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + val withRetransmit = + newEntries.firstOrNull { entry -> + entry.value.tokens.any { it == RecoveryToken.MaxData(maxData = 5_000_000L) } + } + assertNotNull(withRetransmit, "expected re-emit of MaxData(5_000_000)") + assertNull(client.pendingMaxData) + } + + @Test + fun pendingMaxStreamData_perStreamDrain() = + runBlocking { + val client = handshakedClient() + client.lock.lock() + try { + client.pendingMaxStreamData[3L] = 1_024L + client.pendingMaxStreamData[7L] = 2_048L + } finally { + client.lock.unlock() + } + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + val seenTokens = + newEntries + .flatMap { it.value.tokens } + .filterIsInstance() + .toSet() + assertEquals( + setOf( + RecoveryToken.MaxStreamData(streamId = 3L, maxData = 1_024L), + RecoveryToken.MaxStreamData(streamId = 7L, maxData = 2_048L), + ), + seenTokens, + ) + assertTrue(client.pendingMaxStreamData.isEmpty()) + } + + @Test + fun multiplePending_drainEmitsAllInOnePacket(): Unit = + runBlocking { + val client = handshakedClient() + client.lock.lock() + try { + client.pendingMaxStreamsUni = 150L + client.pendingMaxStreamsBidi = 200L + client.pendingMaxData = 1_000_000L + } finally { + client.lock.unlock() + } + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + // All three retransmits should land in one packet (writer + // drains them sequentially before any threshold check). + val packetTokens = + newEntries + .firstOrNull { entry -> + entry.value.tokens.any { it is RecoveryToken.MaxStreamsUni } && + entry.value.tokens.any { it is RecoveryToken.MaxStreamsBidi } && + entry.value.tokens.any { it is RecoveryToken.MaxData } + } + assertNotNull(packetTokens, "expected one packet carrying all three retransmits") + assertNull(client.pendingMaxStreamsUni) + assertNull(client.pendingMaxStreamsBidi) + assertNull(client.pendingMaxData) + } + + @Test + fun noPending_drainDoesNotEmitRetransmits() = + runBlocking { + val client = handshakedClient() + // Nothing to retransmit — no SentPacket should carry a + // MAX_STREAMS / MAX_DATA / MAX_STREAM_DATA token unless + // the rolling-extension threshold check fires (which it + // shouldn't here — fresh handshake, no peer activity). + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + val anyExtensionToken = + newEntries + .flatMap { it.value.tokens } + .any { it is RecoveryToken.MaxStreamsUni || it is RecoveryToken.MaxStreamsBidi || it is RecoveryToken.MaxData || it is RecoveryToken.MaxStreamData } + assertEquals(false, anyExtensionToken, "no pending → no extension tokens") + } + + @Test + fun pendingDrainBeforeThresholdCheck_supersedeOrderObservable(): Unit = + runBlocking { + // Set pendingMaxStreamsUni to a value that's below the current + // advertised cap. The writer drains it as-is — supersede check + // is in step 6 (the setter side), not here. + val client = handshakedClient() + client.lock.lock() + try { + client.pendingMaxStreamsUni = 50L + } finally { + client.lock.unlock() + } + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + val withRetransmit = + newEntries.firstOrNull { entry -> + entry.value.tokens.any { it == RecoveryToken.MaxStreamsUni(maxStreams = 50L) } + } + assertNotNull( + withRetransmit, + "writer must drain pendingMaxStreamsUni unconditionally; supersede check belongs to the setter (step 6)", + ) + } + + @Test + fun pendingClearedAcrossDrains() = + runBlocking { + val client = handshakedClient() + client.lock.lock() + try { + client.pendingMaxStreamsUni = 150L + } finally { + client.lock.unlock() + } + // Drain once: pending consumed. + runCatching { drainOutbound(client, nowMillis = 1L) } + assertNull(client.pendingMaxStreamsUni) + + // Drain again with no new pending: no retransmit token in any + // SentPacket recorded by this drain. + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 2L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + val msuTokens = + newEntries.flatMap { it.value.tokens }.filterIsInstance() + assertEquals(emptyList(), msuTokens, "second drain must not re-emit cleared pending") + } + + private fun handshakedClient(): QuicConnection = + runBlocking { + val client = + QuicConnection( + serverName = "example.test", + config = QuicConnectionConfig(), + 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 + } +}