fix(quic): visibility, scratch caching, retransmit coalescing, secret hygiene

Round 12 — six smaller follow-ups across visibility, perf, and secret-handling. * QuicStream.receiveDirtyForFlowControl gains @Volatile. The parser's read-loop writes the flag and the writer's send-loop reads it WITHOUT holding the same lock; without volatile the writer could miss the parser's update for an unbounded time on JVM (the field is hot in a drain loop where the JIT might cache it), suppressing MAX_STREAM_DATA emissions until something else triggered a fresh load. * SendBuffer.readableBytes runs in O(1) instead of O(R) by maintaining a cached `retransmitTotalBytes` counter. Updated in lockstep with every retransmit deque mutation: addLast in [requeueAllInflight] + the two paths in [removeOverlap] (RETRANSMIT zero-length + main range), and add/removeFirst in [takeChunk]. Pre-flight "anything to send?" check on the writer's hot path was previously walking the deque per-stream per-drain. * SendBuffer.requeueAllInflight coalesces adjacent ranges on insert. Pre-fix the PTO probe path appended each in-flight range as a separate retransmit entry, so on the next drain takeChunk emitted one tiny STREAM frame per original-packet boundary. With coalescing, contiguous bytes get replayed as one chunk + one AEAD seal. FIN-bearing ranges stay separate (merging across a FIN changes the implicit final-size invariant). * TlsResumptionState dropped `data class`. The auto-generated equals/hashCode used reference equality on its ByteArray fields (PSK / ticket / peerTransportParameters), so two byte-identical states compared unequal — almost never useful and a footgun for caller-side caches. The auto-toString would dump PSK contents into any log. Replaced with hand-written equals/hashCode using contentEquals on the byte fields and a redacted toString that reveals only sizes. * QuicConnectionParser RESET_STREAM handler bounds finalSize at [0, 2^62-1] per RFC 9000 §16 (the QUIC offset ceiling). Pre-fix we accepted any varint, including values that could overflow downstream Long math. * PathChallenge/PathResponse IAE leak: re-traced and verified non-issue. The decoder calls `r.readBytes(8)`, which either throws QuicCodecException (short read) or returns exactly 8 bytes — the constructor's `require(data.size == 8)` is unreachable from the decode path. The audit was over-cautious; no code change. All 269 :quic:jvmTest tests pass. BUILD SUCCESSFUL in 39s. https://claude.ai/code/session_01AhGvbMV8uPRse3TmAGaddM
2026-05-09 00:59:28 +00:00
parent e7b7d99582
commit 953869714b
4 changed files with 129 additions and 16 deletions
@@ -67,6 +67,10 @@ import com.vitorpamplona.quic.tls.TlsClient
 * under streamsLock so frame-dispatch / stream creation / level state
 * remains a single critical section.
 */
+
+/** RFC 9000 §16: maximum varint value, also the per-stream offset ceiling. */
+private const val MAX_QUIC_OFFSET: Long = (1L shl 62) - 1L
+
 fun feedDatagram(
    conn: QuicConnection,
    datagram: ByteArray,
@@ -780,9 +784,17 @@ private fun dispatchFrames(
                // RFC 9000 §4.5: the [finalSize] in RESET_STREAM MUST agree
                // with any final size implied by previously-received STREAM
                // frames AND MUST be ≥ the highest offset already observed.
-                // A peer that violates this is closed with FINAL_SIZE_ERROR
-                // — pre-fix we accepted any value silently, letting a buggy
-                // peer drift our state.
+                // §4.5 also caps final size at 2^62-1 (the QUIC offset field
+                // ceiling). A peer that violates any of these is closed with
+                // FINAL_SIZE_ERROR — pre-fix we accepted any value silently,
+                // letting a buggy peer drift our state.
+                if (frame.finalSize < 0L || frame.finalSize > MAX_QUIC_OFFSET) {
+                    conn.markClosedExternally(
+                        "FINAL_SIZE_ERROR: stream ${frame.streamId} RESET_STREAM finalSize " +
+                            "${frame.finalSize} outside [0, 2^62-1]",
+                    )
+                    return
+                }
                val target = conn.streamByIdLocked(frame.streamId)
                if (target != null) {
                    val priorFin = target.receive.finOffset
@@ -130,10 +130,18 @@ class QuicStream(
     * writer's appendFlowControlUpdates consumes it to skip streams that
     * haven't received any new bytes since the last MAX_STREAM_DATA emission.
     *
+     * `@Volatile` because the parser writes it from the read loop and the
+     * writer reads it from the send loop without holding the same lock.
+     * Without volatile the writer could miss the parser's update for an
+     * unbounded time on JVM (the field is read in a hot drain loop where
+     * the JIT might cache it), suppressing MAX_STREAM_DATA emissions
+     * until something else triggered a fresh load.
+     *
     * Pre-fix the writer iterated EVERY open stream on every drain
     * (audit-4 perf #9 — O(streams) × ~50 drains/sec; significant for audio
     * rooms with many WT streams).
     */
+    @Volatile
    internal var receiveDirtyForFlowControl: Boolean = false

    /** True once we've FIN'd our write side and the peer FIN'd theirs. */
@@ -136,6 +136,21 @@ class SendBuffer(
     */
    private val retransmit: ArrayDeque<Range> = ArrayDeque()

+    /**
+     * Cached `sum(retransmit[].length)` so [readableBytes] can return
+     * in O(1) instead of walking the deque on every call. The writer's
+     * pre-flight "anything to send?" check runs per-stream per-drain
+     * (~50 drains/sec × N streams), so an O(R) per-call cost on lossy
+     * paths with deep retransmit queues was meaningful CPU.
+     *
+     * Updated in lockstep with every [retransmit] add/remove path:
+     * retransmit.addLast / addFirst / removeFirst, and
+     * [removeOverlap]'s mid-walk add/remove during ACK processing.
+     * The invariant `retransmitTotalBytes == retransmit.sumOf { it.length }`
+     * holds whenever no [SendBuffer] method is mid-mutation.
+     */
+    private var retransmitTotalBytes: Long = 0L
+
    private var _finPending: Boolean = false
    private var _finSent: Boolean = false
    private var _finAcked: Boolean = false
@@ -154,9 +169,7 @@ class SendBuffer(
    val readableBytes: Int
        get() =
            synchronized(this) {
-                var sum = 0L
-                for (r in retransmit) sum += r.length
-                sum += (_nextOffset - nextSendOffset)
+                val sum = retransmitTotalBytes + (_nextOffset - nextSendOffset)
                sum.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()
            }

@@ -210,16 +223,18 @@ class SendBuffer(
                val take = minOf(retransmitHead.length, cap.toLong())
                val payload = sliceAt(retransmitHead.offset, take.toInt())
                retransmit.removeFirst()
+                retransmitTotalBytes -= retransmitHead.length
                val fin = retransmitHead.fin && take == retransmitHead.length
                if (take < retransmitHead.length) {
                    // Push remainder back at the head, preserving offset.
-                    retransmit.addFirst(
+                    val remainder =
                        Range(
                            offset = retransmitHead.offset + take,
                            length = retransmitHead.length - take,
                            fin = retransmitHead.fin,
-                        ),
-                    )
+                        )
+                    retransmit.addFirst(remainder)
+                    retransmitTotalBytes += remainder.length
                }
                addToInFlight(Range(retransmitHead.offset, take, fin))
                if (fin) _finSent = true
@@ -351,13 +366,29 @@ class SendBuffer(
            }
            // Move every inflight range to the retransmit queue,
            // preserving offset order (inFlight is sorted by offset
-            // ascending, so addLast preserves sort within retransmit
-            // for these new entries — though retransmit is a FIFO
-            // and doesn't strictly require sorted order, takeChunk
-            // pops front-first regardless).
+            // ascending). Coalesce adjacent ranges on insert: if the
+            // previous tail entry's `offset + length` equals the
+            // current's `offset`, merge them into a single range. The
+            // PTO probe path otherwise emits one tiny STREAM frame per
+            // original-packet boundary instead of replaying the
+            // contiguous bytes as a single chunk, fragmenting the
+            // probe across N small frames + N AEAD seals + N writes
+            // when one frame would do. Coalescing across the FIN bit
+            // is gated — only merge when the previous tail had no
+            // FIN, otherwise the FIN's implicit final-size invariant
+            // could shift.
            for (r in inFlight) {
                if (r.fin && !_finAcked) _finSent = false
-                retransmit.addLast(r)
+                val tail = retransmit.lastOrNull()
+                if (tail != null && !tail.fin && tail.offset + tail.length == r.offset) {
+                    // Merge: extend the tail to cover [tail.offset, r.endOffset).
+                    val merged = Range(tail.offset, tail.length + r.length, r.fin)
+                    retransmit.removeLast()
+                    retransmit.addLast(merged)
+                } else {
+                    retransmit.addLast(r)
+                }
+                retransmitTotalBytes += r.length
            }
            inFlight.clear()
        }
@@ -405,6 +436,7 @@ class SendBuffer(

                        OverlapAction.RETRANSMIT -> {
                            retransmit.addLast(r)
+                            retransmitTotalBytes += r.length
                        }

                        OverlapAction.DROP -> {} // discard
@@ -482,6 +514,7 @@ class SendBuffer(
                                fin = coveredFin,
                            ),
                        )
+                        retransmitTotalBytes += coveredLen
                    }

                    OverlapAction.DROP -> {} // discard the covered piece
@@ -677,7 +677,23 @@ interface TlsSecretsListener {
 * [ticketAgeAdd] and [issuedAtMillis] to compute the obfuscated_ticket_age
 * the server expects.
 */
-data class TlsResumptionState(
+/**
+ * Cached state from a prior TLS handshake that lets the client offer
+ * RFC 8446 PSK-resumption + RFC 9001 §4.6 0-RTT on the next connection.
+ *
+ * Plain `class` (not `data class`) intentionally:
+ *  - The `data class`-generated `equals` / `hashCode` use reference
+ *    equality on [ByteArray] fields, so two states with byte-identical
+ *    PSKs would compare unequal — surprising and almost never useful.
+ *  - The auto-generated `toString` would dump `psk` / `ticket` /
+ *    `peerTransportParameters` byte contents into any log, stack trace,
+ *    or debugger that touches the object. Both are sensitive.
+ *
+ * We override [equals] / [hashCode] with [contentEquals] semantics on
+ * the byte fields (so callers can deduplicate cached states by content)
+ * and [toString] to redact the secret payload.
+ */
+class TlsResumptionState(
    /** Opaque ticket bytes echoed verbatim as the PSK identity on the next connection. */
    val ticket: ByteArray,
    /** PSK derived from `resumption_master_secret` + `ticket_nonce` per RFC 8446 §4.6.1. */
@@ -716,7 +732,51 @@ data class TlsResumptionState(
    val peerTransportParameters: ByteArray? = null,
    /** Negotiated ALPN from the prior connection. 0-RTT must use the same protocol. */
    val negotiatedAlpn: ByteArray? = null,
-)
+) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (other !is TlsResumptionState) return false
+        return ticket.contentEquals(other.ticket) &&
+            psk.contentEquals(other.psk) &&
+            cipherSuite == other.cipherSuite &&
+            ticketAgeAdd == other.ticketAgeAdd &&
+            ticketLifetimeSec == other.ticketLifetimeSec &&
+            issuedAtMillis == other.issuedAtMillis &&
+            maxEarlyDataSize == other.maxEarlyDataSize &&
+            (peerTransportParameters?.contentEquals(other.peerTransportParameters) ?: (other.peerTransportParameters == null)) &&
+            (negotiatedAlpn?.contentEquals(other.negotiatedAlpn) ?: (other.negotiatedAlpn == null))
+    }
+
+    override fun hashCode(): Int {
+        var h = ticket.contentHashCode()
+        h = 31 * h + psk.contentHashCode()
+        h = 31 * h + cipherSuite
+        h = 31 * h + ticketAgeAdd.hashCode()
+        h = 31 * h + ticketLifetimeSec.hashCode()
+        h = 31 * h + issuedAtMillis.hashCode()
+        h = 31 * h + maxEarlyDataSize.hashCode()
+        h = 31 * h + (peerTransportParameters?.contentHashCode() ?: 0)
+        h = 31 * h + (negotiatedAlpn?.contentHashCode() ?: 0)
+        return h
+    }
+
+    /**
+     * Redacted [toString]: never include `ticket`, `psk`, or
+     * `peerTransportParameters` byte contents — those leak into logs
+     * and stack traces. Sizes are fine to expose.
+     */
+    override fun toString(): String =
+        "TlsResumptionState(" +
+            "ticket=<${ticket.size} bytes>, " +
+            "psk=<${psk.size} bytes redacted>, " +
+            "cipherSuite=0x${cipherSuite.toString(16)}, " +
+            "ticketAgeAdd=$ticketAgeAdd, " +
+            "ticketLifetimeSec=$ticketLifetimeSec, " +
+            "issuedAtMillis=$issuedAtMillis, " +
+            "maxEarlyDataSize=$maxEarlyDataSize, " +
+            "peerTransportParameters=${peerTransportParameters?.let { "<${it.size} bytes>" }}, " +
+            "negotiatedAlpn=${negotiatedAlpn?.decodeToString()})"
+}

 /** Pluggable certificate validator. Decoupled so we can stub it in tests. */
 interface CertificateValidator {