feat(quic): 0-RTT (early data) — picoquic + quic-go pass

Closes the matrix gap. The TLS layer now drives a full RFC 9001 §4.10
0-RTT path:

- Resumption ClientHello includes the empty `early_data` extension
  when the cached TlsResumptionState carries maxEarlyDataSize > 0
  (parsed from the prior connection's NewSessionTicket early_data
  extension).
- TlsClient.start, post-CH-transcript-snapshot: derive
  client_early_traffic_secret + surface via
  TlsSecretsListener.onEarlyDataKeysReady.
- QuicConnection.zeroRttSendProtection slot installed in the listener
  and cleared in onApplicationKeysReady (RFC 9001 §4.10 forbids 0-RTT
  use after 1-RTT keys are available).
- TlsResumptionState now also carries peerTransportParameters +
  negotiatedAlpn from the issuing connection so a resumed connection
  can pre-load flow-control limits (initial_max_data,
  initial_max_streams_bidi, etc.) BEFORE the new ServerHello arrives.
  Without this, peerMaxStreamsBidi=0 and pre-handshake stream
  creation fails. RFC 9001 §7.4.1 explicitly carves out which
  parameters MUST be remembered for 0-RTT vs which MUST NOT (CIDs,
  ack delay).
- QuicConnectionWriter.buildApplicationPacket: dual 0-RTT / 1-RTT
  path. When 1-RTT keys are absent but 0-RTT keys are present, build
  a long-header type=0x01 ZERO_RTT packet (sharing the Application
  packet number space per RFC 9000 §17.2.3) and skip ACK frames
  (server cannot ACK 0-RTT-level packets). Once 1-RTT installs, the
  writer naturally falls through to short-header.
- InteropClient runResumptionTest gains a `zerortt` flag. When set,
  iter 0 fetches NOTHING (just establishes + waits the existing
  200ms post-handshake window for the NewSessionTicket to arrive +
  closes), and iter 1 opens all URLs as bidi streams + enqueues GETs
  + driver.wakeup BEFORE awaitHandshake so the writer ships them as
  0-RTT packets coalesced with (or right after) the resumed
  ClientHello in the first datagram.

Results:
- ✓ picoquic: 0-RTT 10682 bytes, 1-RTT 238 bytes — within the
  runner's 50% / 5000-byte 1-RTT cap.
- ✓ quic-go: 0-RTT 10693 bytes, 1-RTT 1488 bytes — same.
- ✕ aioquic: server rejects our 0-RTT (only 3 STREAM frames come
  back from 40 GETs sent); no rejection-fallback wired (a real
  implementation would track which app data was sent in 0-RTT and
  replay in 1-RTT after EE comes back without early_data
  acceptance). Out of scope for this pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt

@@ -34,6 +34,7 @@ import kotlinx.coroutines.async
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeoutOrNull
 import java.io.File
@@ -275,7 +276,27 @@ fun main() {
             // the PSK already authenticated us). Different shape from
             // multiconnect (which never resumes) so it has its own
             // dispatch.
-            "resumption" -> {
+            "resumption", "zerortt" -> {
+                // zerortt extends resumption — the runner's testcase
+                // verifies the pcap shows >0 bytes in 0-RTT packets and
+                // ≤50% of total client-direction bytes in 1-RTT. Same
+                // 2-connection shape: connection 1 captures the ticket,
+                // connection 2 resumes. The 0-RTT path activates when
+                // the captured ticket has maxEarlyDataSize > 0 (set
+                // by the issuing server's NewSessionTicket early_data
+                // extension).
+                //
+                // Difference between the two testcases is the URL split:
+                //
+                //  - resumption: split URLs in half across the two
+                //    connections so the runner sees 2 distinct
+                //    handshakes both transferring data.
+                //  - zerortt: connection 1 fetches NOTHING (just hold
+                //    the conn open long enough to receive the
+                //    NewSessionTicket), connection 2 fetches ALL URLs
+                //    via 0-RTT. This keeps iter 0's 1-RTT byte count at
+                //    ~zero so the runner's 50% cap on 1-RTT bytes
+                //    stays comfortably under the limit.
                 runResumptionTest(
                     requests = requests,
                     downloadsDir = downloadsDir,
@@ -284,6 +305,7 @@ fun main() {
                     initialVersion = initialVersion,
                     keyLogPath = keyLogPath,
                     qlogDir = qlogDir,
+                    zerortt = (testcase == "zerortt"),
                 )
             }
 
@@ -681,6 +703,8 @@ private fun runResumptionTest(
     initialVersion: Int,
     keyLogPath: String?,
     qlogDir: File?,
+    /** When true, route ALL URLs to the second (resumed) connection's 0-RTT path. */
+    zerortt: Boolean = false,
 ): Int {
     val urls =
         requests
@@ -698,11 +722,26 @@ private fun runResumptionTest(
     qlogDir?.mkdirs()
     val keyLogger = keyLogPath?.let { SslKeyLogger(File(it)) }
 
-    // Split request list in half: first half on the full-handshake
+    // resumption: split URLs in half across the two connections.
-    // connection, second half on the resumed PSK connection.
+    // zerortt: connection 1 fetches nothing (just gets the
-    val splitAt = urls.size / 2
+    // NewSessionTicket), connection 2 fetches all URLs via 0-RTT to
-    val firstUrls = urls.subList(0, splitAt.coerceAtLeast(1))
+    // keep the 1-RTT byte count from iter 0 GETs out of the runner's
-    val secondUrls = urls.subList(splitAt.coerceAtLeast(1), urls.size)
+    // 1-RTT cap.
+    val firstUrls: List<URI>
+    val secondUrls: List<URI>
+    if (zerortt) {
+        firstUrls = emptyList()
+        secondUrls = urls
+    } else {
+        val splitAt = urls.size / 2
+        firstUrls = urls.subList(0, splitAt.coerceAtLeast(1))
+        secondUrls = urls.subList(splitAt.coerceAtLeast(1), urls.size)
+    }
+
+    // For the zerortt no-data iter 0 we still need a host/port to
+    // connect to — borrow the first URL purely for its authority.
+    val firstUrlsForConnection = if (firstUrls.isEmpty()) urls.subList(0, 1) else firstUrls
+    val firstUrlsForGet = firstUrls
 
     val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
     val outcome =
@@ -712,7 +751,8 @@ private fun runResumptionTest(
             val outcome1 =
                 runOneResumptionConnection(
                     iterIdx = 0,
-                    urls = firstUrls,
+                    urls = firstUrlsForConnection,
+                    fetchUrls = firstUrlsForGet,
                     downloadsDir = downloadsDir,
                     cipherSuites = cipherSuites,
                     offeredAlpns = offeredAlpns,
@@ -736,6 +776,7 @@ private fun runResumptionTest(
                 runOneResumptionConnection(
                     iterIdx = 1,
                     urls = secondUrls,
+                    fetchUrls = secondUrls,
                     downloadsDir = downloadsDir,
                     cipherSuites = cipherSuites,
                     offeredAlpns = offeredAlpns,
@@ -761,7 +802,15 @@ private fun runResumptionTest(
 
 private suspend fun runOneResumptionConnection(
     iterIdx: Int,
+    /** URLs whose authority drives socket connect — must be non-empty. */
     urls: List<URI>,
+    /**
+     * URLs to actually fetch. Distinct from [urls] for the zerortt
+     * iter 0 case where we want to establish a connection (using the
+     * first URL's authority) but not GET anything — the iter exists
+     * solely to receive a NewSessionTicket the next iter can resume on.
+     */
+    fetchUrls: List<URI>,
     downloadsDir: File,
     cipherSuites: IntArray?,
     offeredAlpns: List<Alpn>,
@@ -814,6 +863,40 @@ private suspend fun runOneResumptionConnection(
     val driver = QuicConnectionDriver(conn, socket, scope)
     driver.start()
 
+    // 0-RTT path: when this iteration is on a resumed connection AND the
+    // resumption ticket allowed early data, open the bidi streams and
+    // enqueue the GET requests BEFORE awaiting the handshake. The
+    // writer will pick them up using the 0-RTT keys derived in
+    // onEarlyDataKeysReady (long-header type=0x01) and ship them
+    // alongside (or right after) the ClientHello in the first
+    // datagram. RFC 9001 §4.6 / RFC 8446 §4.2.10 — the server may
+    // accept or reject 0-RTT; if rejected the data is silently lost
+    // on the server side and we'd need to re-send post-handshake. For
+    // the runner's 0-RTT testcase against a server that accepts, we
+    // get away without the rebuild — the server processes the
+    // requests and replies in 1-RTT after handshake.
+    val zeroRttPlanned = resumption != null && resumption.maxEarlyDataSize > 0 && fetchUrls.isNotEmpty()
+    val pre0RttHandles =
+        if (zeroRttPlanned) {
+            // ALPN isn't yet renegotiated for THIS connection (we're
+            // pre-handshake), so use the cached ALPN from the prior
+            // connection — RFC 9001 §4.6 requires the same ALPN when
+            // sending 0-RTT data.
+            val cachedAlpn = resumption!!.negotiatedAlpn?.decodeToString().orEmpty()
+            // Pre-handshake stream open works because QuicConnection.init
+            // pre-loaded peerMaxStreamsBidi from the resumption state.
+            val handles =
+                conn.openBidiStreamsBatch(fetchUrls.map { "GET ${it.path}\r\n".encodeToByteArray() }) { stream, request ->
+                    stream.send.enqueue(request)
+                    stream.send.finish()
+                    stream
+                }
+            driver.wakeup()
+            cachedAlpn to handles
+        } else {
+            null
+        }
+
     val handshake =
         withTimeoutOrNull(HANDSHAKE_TIMEOUT_SEC * 1_000L) {
             runCatching { conn.awaitHandshake() }
@@ -846,23 +929,55 @@ private suspend fun runOneResumptionConnection(
         }
 
     var anyFailed = false
-    for (url in urls) {
+    if (pre0RttHandles != null) {
-        val resp =
+        // 0-RTT path: streams already opened and GETs already enqueued
-            withTimeoutOrNull(TRANSFER_TIMEOUT_SEC * 1_000L) {
+        // pre-handshake. Just collect the responses on each stream.
-                client.get(authority, url.path)
+        // Server may have accepted the 0-RTT data (responses come back)
+        // or rejected it; rejection means data was dropped server-side
+        // and we'd have to resend in 1-RTT, which is real work and not
+        // wired here. For the runner's zerortt testcase against picoquic
+        // (which accepts), this path is sufficient.
+        val (_, handles) = pre0RttHandles
+        for ((url, stream) in fetchUrls.zip(handles)) {
+            val body =
+                withTimeoutOrNull(TRANSFER_TIMEOUT_SEC * 1_000L) {
+                    val chunks = stream.incoming.toList()
+                    val total = chunks.sumOf { it.size }
+                    val buf = ByteArray(total)
+                    var off = 0
+                    for (c in chunks) {
+                        c.copyInto(buf, off)
+                        off += c.size
+                    }
+                    buf
+                }
+            if (body == null || body.isEmpty()) {
+                anyFailed = true
+                System.err.println("[resumption:$iterIdx] 0RTT GET ${url.path} → empty/timeout")
+                continue
             }
-        if (resp == null) {
+            val name = url.path.substringAfterLast('/').ifBlank { "index" }
-            anyFailed = true
+            File(downloadsDir, name).writeBytes(body)
-            System.err.println("[resumption:$iterIdx] GET ${url.path} → timeout")
-            break
         }
-        if (resp.status != 200) {
+    } else {
-            System.err.println("[resumption:$iterIdx] GET ${url.path} → status ${resp.status}")
+        for (url in fetchUrls) {
-            anyFailed = true
+            val resp =
-            continue
+                withTimeoutOrNull(TRANSFER_TIMEOUT_SEC * 1_000L) {
+                    client.get(authority, url.path)
+                }
+            if (resp == null) {
+                anyFailed = true
+                System.err.println("[resumption:$iterIdx] GET ${url.path} → timeout")
+                break
+            }
+            if (resp.status != 200) {
+                System.err.println("[resumption:$iterIdx] GET ${url.path} → status ${resp.status}")
+                anyFailed = true
+                continue
+            }
+            val name = url.path.substringAfterLast('/').ifBlank { "index" }
+            File(downloadsDir, name).writeBytes(resp.body)
         }
-        val name = url.path.substringAfterLast('/').ifBlank { "index" }
-        File(downloadsDir, name).writeBytes(resp.body)
     }
 
     // Give the server a chance to send its NewSessionTicket — picoquic

quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt

@@ -610,6 +610,30 @@ class QuicConnection(
             PacketProtection(bestAes128GcmAead(proto.clientKey), proto.clientKey, proto.clientIv, hp, proto.clientHp)
         initial.receiveProtection =
             PacketProtection(bestAes128GcmAead(proto.serverKey), proto.serverKey, proto.serverIv, hp, proto.serverHp)
+
+        // Resumption + 0-RTT: pre-load flow-control limits from the
+        // remembered transport parameters so streams opened BEFORE the
+        // new connection's ServerHello arrives have credit to push 0-RTT
+        // STREAM frames on the wire. RFC 9001 §7.4.1 explicitly allows
+        // (in fact requires) the client to use REMEMBERED transport
+        // params for this purpose, while skipping the CID-bound ones
+        // (initial_source_connection_id etc.) which would mismatch the
+        // fresh CIDs we just generated. Server's real new params arrive
+        // in EncryptedExtensions and the existing
+        // [applyPeerTransportParameters] hook then overwrites these
+        // pre-loaded values.
+        if (resumption?.peerTransportParameters != null) {
+            try {
+                val tp = TransportParameters.decode(resumption.peerTransportParameters)
+                sendConnectionFlowCredit = tp.initialMaxData ?: 0L
+                peerMaxStreamsBidi = tp.initialMaxStreamsBidi ?: 0L
+                peerMaxStreamsUni = tp.initialMaxStreamsUni ?: 0L
+            } catch (_: Throwable) {
+                // Bad cached params shouldn't block the connection; we
+                // just won't be able to send 0-RTT data. Server's
+                // real params arrive on EE either way.
+            }
+        }
     }
 
     /** Begin the handshake — emits ClientHello into Initial CRYPTO. */

quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt

@@ -568,7 +568,13 @@ private fun buildApplicationPacket(
     nowMillis: Long,
 ): ByteArray? {
     val state = conn.application
-    val proto = state.sendProtection ?: return null
+    // Prefer 1-RTT keys when available; fall back to 0-RTT keys for the
+    // pre-handshake-confirmed window on a resumption connection. Once
+    // 1-RTT lands, [QuicConnection.zeroRttSendProtection] is cleared per
+    // RFC 9001 §4.10 — so the fallback only ever fires before
+    // ServerHello has been processed.
+    val use1Rtt = state.sendProtection != null
+    val proto = state.sendProtection ?: conn.zeroRttSendProtection ?: return null
     val frames = mutableListOf<Frame>()
     // Tokens collected in lock-step with [frames]: each retransmittable
     // frame contributes a [RecoveryToken] so the [SentPacket] recorded
@@ -580,32 +586,39 @@ private fun buildApplicationPacket(
     // tracked for loss-detection timing.
     val tokens = mutableListOf<RecoveryToken>()
 
-    state.ackTracker.buildAckFrame(nowMillis, conn.config.ackDelayExponent.toInt())?.let { plainAck ->
+    // RFC 9000 §17.2.3 — 0-RTT packets MUST NOT contain ACK frames. The
-        // RFC 9000 §13.4.2: an endpoint that USES ECN on outbound
+    // server cannot ACK 0-RTT-level packets because it has no way to
-        // packets (we set ECT(0) on every datagram via the socket's
+    // signal which encryption level the ACK targets without the
-        // IP_TOS option) MUST report ECN counts in 1-RTT ACK frames so
+    // application packet number space being established by 1-RTT keys.
-        // the peer can detect path congestion. We don't currently
+    // Skip ACK building when we're about to emit a 0-RTT packet.
-        // read inbound TOS bits — JDK's DatagramChannel doesn't expose
+    if (use1Rtt) {
-        // them without JNI — so the counts are all-zero. The interop
+        state.ackTracker.buildAckFrame(nowMillis, conn.config.ackDelayExponent.toInt())?.let { plainAck ->
-        // runner's `ecn` testcase only checks for the field's presence
+            // RFC 9000 §13.4.2: an endpoint that USES ECN on outbound
-        // (`hasattr(p["quic"], "ack.ect0_count")`); strict peers that
+            // packets (we set ECT(0) on every datagram via the socket's
-        // cross-validate counts would reject this, but aioquic /
+            // IP_TOS option) MUST report ECN counts in 1-RTT ACK frames so
-        // picoquic / quic-go all tolerate it. Initial / Handshake-space
+            // the peer can detect path congestion. We don't currently
-        // ACKs stay plain (ecnCounts=null) — the spec allows ECN counts
+            // read inbound TOS bits — JDK's DatagramChannel doesn't expose
-        // there too, but interop implementations don't always handle
+            // them without JNI — so the counts are all-zero. The interop
-        // them and we'd gain nothing.
+            // runner's `ecn` testcase only checks for the field's presence
-        val ackWithEcn =
+            // (`hasattr(p["quic"], "ack.ect0_count")`); strict peers that
-            AckFrame(
+            // cross-validate counts would reject this, but aioquic /
-                largestAcknowledged = plainAck.largestAcknowledged,
+            // picoquic / quic-go all tolerate it. Initial / Handshake-space
-                ackDelay = plainAck.ackDelay,
+            // ACKs stay plain (ecnCounts=null) — the spec allows ECN counts
-                firstAckRange = plainAck.firstAckRange,
+            // there too, but interop implementations don't always handle
-                additionalRanges = plainAck.additionalRanges,
+            // them and we'd gain nothing.
-                ecnCounts =
+            val ackWithEcn =
-                    com.vitorpamplona.quic.frame
+                AckFrame(
-                        .AckEcnCounts(0L, 0L, 0L),
+                    largestAcknowledged = plainAck.largestAcknowledged,
-            )
+                    ackDelay = plainAck.ackDelay,
-        frames += ackWithEcn
+                    firstAckRange = plainAck.firstAckRange,
-        tokens += RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = ackWithEcn.largestAcknowledged)
+                    additionalRanges = plainAck.additionalRanges,
+                    ecnCounts =
+                        com.vitorpamplona.quic.frame
+                            .AckEcnCounts(0L, 0L, 0L),
+                )
+            frames += ackWithEcn
+            tokens += RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = ackWithEcn.largestAcknowledged)
+        }
     }
 
     // Step 7: PTO probe. The driver sets `pendingPing` when its
@@ -763,20 +776,43 @@ private fun buildApplicationPacket(
     // change visible to the peer.
     val sizeBytes =
         runCatching {
-            ShortHeaderPacket.build(
+            if (use1Rtt) {
-                ShortHeaderPlaintextPacket(
+                ShortHeaderPacket.build(
-                    conn.destinationConnectionId,
+                    ShortHeaderPlaintextPacket(
-                    pn,
+                        conn.destinationConnectionId,
-                    payload,
+                        pn,
-                    keyPhase = conn.currentSendKeyPhase,
+                        payload,
-                ),
+                        keyPhase = conn.currentSendKeyPhase,
-                proto.aead,
+                    ),
-                proto.key,
+                    proto.aead,
-                proto.iv,
+                    proto.key,
-                proto.hp,
+                    proto.iv,
-                proto.hpKey,
+                    proto.hp,
-                largestAckedInSpace = -1L,
+                    proto.hpKey,
-            )
+                    largestAckedInSpace = -1L,
+                )
+            } else {
+                // 0-RTT — long header type=0x01. Same Application packet
+                // number space (RFC 9000 §17.2.3), same DCID/SCID. Token
+                // is empty (only Initial carries a token) — the
+                // LongHeaderPacket builder special-cases that.
+                LongHeaderPacket.build(
+                    LongHeaderPlaintextPacket(
+                        type = LongHeaderType.ZERO_RTT,
+                        version = conn.currentVersion,
+                        dcid = conn.destinationConnectionId,
+                        scid = conn.sourceConnectionId,
+                        packetNumber = pn,
+                        payload = payload,
+                    ),
+                    proto.aead,
+                    proto.key,
+                    proto.iv,
+                    proto.hp,
+                    proto.hpKey,
+                    largestAckedInSpace = -1L,
+                )
+            }
         }
     state.sentPackets[pn] =
         SentPacket(