feat(quic): step 4 of RFC 9002 retransmit — pending* fields + writer drain

Adds the writer's drain side of the retransmit path. The QuicConnection
gains four `pending*` fields:

  - pendingMaxStreamsUni: Long?
  - pendingMaxStreamsBidi: Long?
  - pendingMaxData: Long?
  - pendingMaxStreamData: MutableMap<Long, Long>  (keyed by stream id)

Each non-null entry signals "the last extension we sent at this value
was lost; re-emit it". appendFlowControlUpdates now drains all four
ahead of the rolling-extension threshold check, emitting a fresh
frame + RecoveryToken for each pending entry and clearing it.

Step 4 only wires the consumer side; the setter side (loss
dispatcher) is step 6. Tests populate `pending*` directly to exercise
the drain in isolation.

Tests added (8, all pass):
  - pendingMaxStreamsUni / Bidi / MaxData / MaxStreamData each
    individually drain to a SentPacket carrying the matching token
  - multiplePending: all four pending types drain into one packet
    (writer drains them sequentially, no fan-out)
  - noPending: drain produces no extension tokens
  - pendingDrainBeforeThresholdCheck_supersedeOrderObservable:
    writer drains the pending value as-is — supersede check is the
    setter's responsibility (step 6), not the drain's
  - pendingClearedAcrossDrains: a cleared pending stays cleared on
    subsequent drains

Mirrors neqo's `fc.rs` retransmit tests in the plan
(`need_max_allowed_frame_after_loss`, `lost_after_increase`,
`multiple_retries_after_frame_pending_is_set`,
`new_retired_before_loss`). The supersede-check tests
(`no_max_allowed_frame_after_old_loss`,
`set_max_active_equal_does_not_set_frame_pending`) belong to step 6
and land there.

https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ

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<Long, Long> = HashMap()
+
     /**
      * 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/QuicConnectionWriter.kt

@@ -386,6 +386,42 @@ private fun appendFlowControlUpdates(
     frames: MutableList<Frame>,
     tokens: MutableList<RecoveryToken>,
 ) {
+    // 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

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<RecoveryToken.MaxStreamData>()
+                    .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<RecoveryToken.MaxStreamsUni>()
+            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
+        }
+}