Merge pull request #2802 from vitorpamplona/claude/review-quic-module-0vyTz

Security hardening and performance improvements across QUIC stack

quic/src/commonMain/kotlin/com/vitorpamplona/quic/Buffer.kt

@@ -203,6 +203,9 @@ class QuicReader(
     }
 
     fun skip(n: Int) {
+        if (n < 0) {
+            throw QuicCodecException("skip with negative count: $n")
+        }
         require(n)
         pos += n
     }
@@ -248,6 +251,14 @@ class QuicReader(
     }
 
     fun readBytes(n: Int): ByteArray {
+        // Translate a negative `n` (e.g. an attacker-controlled length
+        // that wrapped through `.toInt()`) into a typed
+        // [QuicCodecException] instead of `IllegalArgumentException`
+        // from `copyOfRange`. Bounds-checking via [require] still kicks
+        // in for positive `n` past `end`.
+        if (n < 0) {
+            throw QuicCodecException("readBytes with negative count: $n")
+        }
         require(n)
         val out = src.copyOfRange(pos, pos + n)
         pos += n
@@ -280,3 +291,17 @@ class QuicCodecException(
     message: String,
     cause: Throwable? = null,
 ) : RuntimeException(message, cause)
+
+/**
+ * Peer protocol violation that mandates connection close per RFC 9000 / 9001.
+ * Distinct from [QuicCodecException] (which is also "drop the packet" for
+ * AEAD-failed inputs) — a [QuicProtocolViolationException] means the peer
+ * sent something well-formed enough to AEAD-decrypt but inconsistent with
+ * the wire spec, so the connection MUST be closed with PROTOCOL_VIOLATION.
+ *
+ * Typical sources: reserved-bit-set in the unmasked QUIC header
+ * (RFC 9000 §17.2 / §17.3.1).
+ */
+class QuicProtocolViolationException(
+    message: String,
+) : RuntimeException(message)

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

@@ -430,7 +430,15 @@ class PathValidator(
         currentPtoMillis: Long,
     ): PathMigrationResult {
         if (state is PathValidationState.Validating) return PathMigrationResult.AlreadyInProgress
-        val (seq, entry) = unusedCids.entries.firstOrNull() ?: return PathMigrationResult.NoSpareCid
+        // Pick the SMALLEST sequence number, not insertion order. The
+        // peer is allowed (RFC 9000 §19.15) to issue NEW_CONNECTION_ID
+        // out of sequence — e.g. retransmits arriving after newer
+        // higher-seq offers. LinkedHashMap iteration is by insertion,
+        // so the prior `firstOrNull` would pick whichever offer landed
+        // first, not the lowest seq. Picking the smallest preserves
+        // RFC's expected ordering (lower seq retired first) and lines
+        // up with what other clients (quicly, neqo) do.
+        val (seq, entry) = unusedCids.entries.minByOrNull { it.key } ?: return PathMigrationResult.NoSpareCid
         unusedCids.remove(seq)
         val payload =
             challengePayloadFactory().also {
@@ -580,7 +588,16 @@ class PathValidator(
      */
     fun forceRotateToHigherSequence(): ForcedRotationResult? {
         if (activeCidSequence >= retirePriorToWatermark) return null
-        val (seq, entry) = unusedCids.entries.firstOrNull() ?: return ForcedRotationResult.NoSpareCid
+        // Pick the smallest seq above the watermark — see
+        // [tryStartValidation]'s rationale. LinkedHashMap is insertion-
+        // ordered, not seq-ordered, so a peer that retransmits an old
+        // offer can shift the "first" entry away from the actual
+        // smallest.
+        val (seq, entry) =
+            unusedCids.entries
+                .filter { it.key >= retirePriorToWatermark }
+                .minByOrNull { it.key }
+                ?: return ForcedRotationResult.NoSpareCid
         unusedCids.remove(seq)
         val priorSeq = activeCidSequence
         queueRetireSequence(priorSeq)

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

@@ -69,11 +69,19 @@ class QuicConnection(
      * of letting null silently disable MITM protection.
      */
     val tlsCertificateValidator: com.vitorpamplona.quic.tls.CertificateValidator,
-    val nowMillis: () -> Long = {
-        kotlin.time.Clock.System
-            .now()
-            .toEpochMilliseconds()
-    },
+    /**
+     * Monotonic clock used by ACK-delay encoding, RTT samples, PTO scheduling,
+     * loss detection, and path-validation timeouts. Default uses
+     * [kotlin.time.TimeSource.Monotonic] so an NTP step / suspend-resume
+     * doesn't poison RTT estimates or trigger spurious losses. The returned
+     * value is "milliseconds since this connection was constructed" — only
+     * differences are meaningful.
+     *
+     * Tests inject a virtual clock; production callers should leave the
+     * default unless they have a specific reason (e.g. recording wallclock
+     * timestamps in qlog) and understand the wallclock pitfalls.
+     */
+    val nowMillis: () -> Long = defaultMonotonicNowMillis(),
     val alpnList: List<ByteArray> = listOf(TlsConstants.ALPN_H3),
     /**
      * Optional second listener invoked after the connection's own
@@ -283,13 +291,27 @@ class QuicConnection(
     /**
      * Lock-split refactor (2026-05-08): @Volatile so concurrent loops can
      * read the status without a lock — coarse "are we still alive?" checks.
-     * Mutating transitions still go through [lifecycleLock] for atomicity
+     * Mutating transitions still go through [closeStateMonitor] for atomicity
      * with [closeReason]/[closeErrorCode] updates.
      */
     @Volatile
     var status: Status = Status.HANDSHAKING
         internal set
 
+    /**
+     * Non-suspend monitor protecting the atomic transition of
+     * [status] / [closeReason] / [closeErrorCode] from a non-CLOSED to a
+     * CLOSED/CLOSING state. We intentionally do NOT use [lifecycleLock]
+     * for this — that's a `kotlinx.coroutines.sync.Mutex` which only
+     * works from suspend contexts, and [markClosedExternally] is invoked
+     * from the parser inside a `streamsLock.withLock { }` block where
+     * suspending again on a different mutex is awkward and risks lock
+     * inversion. A plain monitor avoids both problems while still
+     * giving us a true compare-and-set for "first caller wins the
+     * close-reason and gets to fire the qlog event".
+     */
+    private val closeStateMonitor = Any()
+
     /** App-level error code for graceful close. */
     var closeReason: String? = null
         private set
@@ -299,27 +321,36 @@ class QuicConnection(
     private val streams = mutableMapOf<Long, QuicStream>()
 
     /**
-     * Round-4 perf #10: parallel insertion-ordered list of streams so the
-     * writer's round-robin scan can index by position without
-     * `streams.entries.toList()` allocating per drain. Mutated in lockstep
-     * with [streams] under [streamsLock]:
+     * Per-connection insertion-ordered list of live streams. Mutated only
+     * under [streamsLock] (single-writer):
      *  - [openBidiStreamLocked] / [openUniStreamLocked] /
      *    [getOrCreatePeerStreamLocked] append.
      *  - [retireFullyDoneStreamsLocked] removes entries whose stream has
-     *    flipped [QuicStream.isFullyRetired] = true. The retire pass
-     *    folds the receive-side high-water mark into
-     *    [retiredStreamsRecvBytes] so the writer's MAX_DATA accounting
-     *    in `appendFlowControlUpdates` doesn't regress when a retired
-     *    stream's `receive.contiguousEnd()` drops out of the iteration.
+     *    flipped [QuicStream.isFullyRetired] = true, folding the
+     *    receive-side high-water mark into [retiredStreamsRecvBytes] so
+     *    the writer's MAX_DATA accounting in `appendFlowControlUpdates`
+     *    doesn't regress when a retired stream's `receive.contiguousEnd()`
+     *    drops out of the iteration.
+     *
+     * Read by both lock-holding paths (writer drain, parser dispatch) and
+     * by lock-free observers ([closeAllSignals] after teardown). To make
+     * lock-free reads safe we use an immutable snapshot pattern: every
+     * mutation publishes a fresh `List` via the `@Volatile` reference,
+     * and readers see either the pre- or post-mutation snapshot — never
+     * a half-modified ArrayList that can raise
+     * `ConcurrentModificationException`. The cost is one shallow copy
+     * per add / retire; steady-state churn stays under ~100/sec even at
+     * peak audio-room load (see
+     * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`,
+     * which pegs ~50 streams/sec at peak).
      *
      * Without retirement, the moq-lite audio-rooms path leaks one
-     * QuicStream per Opus frame for the lifetime of the session — the
-     * stream-cliff investigation in
-     * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`
-     * pegs steady-state churn at ~50 streams/sec, so a 3-hour room
-     * accumulates ~540 000 entries before retirement was wired.
+     * QuicStream per Opus frame for the lifetime of the session — a
+     * 3-hour room accumulates ~540 000 entries before retirement was
+     * wired.
      */
-    private val streamsList = mutableListOf<QuicStream>()
+    @Volatile
+    private var streamsList: List<QuicStream> = emptyList()
     private var nextLocalBidiIndex: Long = 0L
     private var nextLocalUniIndex: Long = 0L
 
@@ -926,6 +957,15 @@ class QuicConnection(
         originalPacketBytes: ByteArray,
     ): Boolean {
         if (retryConsumed) return false
+        // RFC 9000 §17.2.5.2: "the client MUST discard a Retry packet
+        // that contains a Source Connection ID field that is identical
+        // to the Destination Connection ID field of its Initial packet".
+        // Without this guard a self-loop / off-path attacker could feed
+        // us a Retry that nominally validates but pins our state to the
+        // attacker's chosen DCID forever.
+        if (retryPacket.scid.bytes.contentEquals(originalDestinationConnectionId.bytes)) {
+            return false
+        }
         if (!retryPacket.verifyIntegrityTag(originalPacketBytes, originalDestinationConnectionId.bytes)) {
             return false
         }
@@ -1184,7 +1224,7 @@ class QuicConnection(
         stream.sendCredit = peerTransportParameters?.initialMaxStreamDataBidiRemote ?: config.initialMaxStreamDataBidiRemote
         stream.receiveLimit = config.initialMaxStreamDataBidiLocal
         streams[id] = stream
-        streamsList += stream
+        streamsList = streamsList + stream
         return stream
     }
 
@@ -1218,7 +1258,7 @@ class QuicConnection(
         stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni
         stream.receiveLimit = 0L // can't receive
         streams[id] = stream
-        streamsList += stream
+        streamsList = streamsList + stream
         return stream
     }
 
@@ -1412,15 +1452,21 @@ class QuicConnection(
         errorCode: Long,
         reason: String,
     ) {
-        var firedQlog = false
-        lifecycleLock.withLock {
-            if (status == Status.CLOSED || status == Status.CLOSING) return@withLock
-            closeErrorCode = errorCode
-            closeReason = reason
-            status = Status.CLOSING
-            firedQlog = true
-        }
-        if (firedQlog) qlogObserver.onConnectionClosed("local", errorCode, reason)
+        // Atomic CAS via [closeStateMonitor] so two concurrent
+        // close()/markClosedExternally callers can't both observe a
+        // non-CLOSED state and both proceed to fire the qlog event /
+        // overwrite [closeReason]. First caller wins; subsequent
+        // callers no-op silently.
+        val firedQlog =
+            synchronized(closeStateMonitor) {
+                if (status == Status.CLOSED || status == Status.CLOSING) return@synchronized false
+                closeErrorCode = errorCode
+                closeReason = reason
+                status = Status.CLOSING
+                true
+            }
+        if (!firedQlog) return
+        qlogObserver.onConnectionClosed("local", errorCode, reason)
         // If a caller is suspended on awaitHandshake() and we're tearing down
         // before completion, fail the deferred so the caller throws instead
         // of hanging forever.
@@ -1432,17 +1478,23 @@ class QuicConnection(
 
     /** Called by the parser on inbound CONNECTION_CLOSE or by the driver on read-loop death. */
     internal fun markClosedExternally(reason: String) {
-        val wasClosed = status == Status.CLOSED
-        if (status != Status.CLOSED) status = Status.CLOSED
-        if (!wasClosed) {
-            // First-call wins for [closeReason] so the highest-quality
-            // diagnostic is preserved when several teardown paths race
-            // (e.g. read loop's `socket.receive() == null` finally fires
-            // a moment before the send loop's `socket.send` throw catch
-            // block does). Without this, downstream observers like
-            // `ReconnectingNestsListener.terminalAwait` see a closed
-            // connection but no human-readable cause for the failure.
-            closeReason = reason
+        // First-call wins for [closeReason] so the highest-quality
+        // diagnostic is preserved when several teardown paths race
+        // (e.g. read loop's `socket.receive() == null` finally fires
+        // a moment before the send loop's `socket.send` throw catch
+        // block does). Pre-fix, the "did we win the race?" check was a
+        // non-atomic read-then-write on [status], so two callers could
+        // both observe `status != CLOSED`, both fire the qlog event,
+        // and both stomp on [closeReason] in unpredictable order. The
+        // monitor below makes the transition truly atomic.
+        val firstClose =
+            synchronized(closeStateMonitor) {
+                if (status == Status.CLOSED) return@synchronized false
+                status = Status.CLOSED
+                closeReason = reason
+                true
+            }
+        if (firstClose) {
             // "remote" covers both peer-initiated CONNECTION_CLOSE and
             // local invariant violations (CID mismatch, frame decode
             // failure) that the parser surfaces as markClosedExternally.
@@ -1480,7 +1532,12 @@ class QuicConnection(
         closedSignal.close()
         peerStreamSignal.close()
         incomingDatagramSignal.close()
-        // Iterate the snapshot list (safe: we never remove from it).
+        // [streamsList] is now an immutable List published via @Volatile,
+        // so reading it here without [streamsLock] yields a consistent
+        // snapshot — either the pre-mutation or post-mutation view —
+        // and never a half-mutated ArrayList raising CME. Mutators
+        // (open*Locked, retireFullyDoneStreamsLocked) all run under
+        // [streamsLock] and publish a fresh List on each change.
         // closeIncoming is idempotent on the underlying Channel.close().
         for (stream in streamsList) {
             stream.closeIncoming()
@@ -1527,7 +1584,7 @@ class QuicConnection(
                 StreamId.Kind.CLIENT_BIDI -> config.initialMaxStreamDataBidiLocal
             }
         streams[id] = stream
-        streamsList += stream
+        streamsList = streamsList + stream
         newPeerStreams.addLast(stream)
         // Track lifetime peer-stream counts so the writer can emit a
         // refreshed MAX_STREAMS_* once the peer's usage approaches the
@@ -1882,12 +1939,20 @@ class QuicConnection(
      *    stream that delivers duplicate bytes to the application.
      */
     internal fun retireFullyDoneStreamsLocked(): Int {
-        if (streamsList.isEmpty()) return 0
+        val current = streamsList
+        if (current.isEmpty()) return 0
+        // Build the post-retire snapshot in one pass. Mutating the live
+        // [streamsList] in-place would force readers (closeAllSignals,
+        // diagnostic accessors) to take [streamsLock] just to iterate;
+        // building a fresh list and publishing it via the @Volatile ref
+        // makes those reads lock-free and CME-free.
         var removed = 0
-        val it = streamsList.iterator()
-        while (it.hasNext()) {
-            val stream = it.next()
-            if (!stream.isFullyRetired) continue
+        val kept = ArrayList<QuicStream>(current.size)
+        for (stream in current) {
+            if (!stream.isFullyRetired) {
+                kept.add(stream)
+                continue
+            }
             // Fold the per-stream receive high-water into the cumulative
             // counter BEFORE we drop it — once we lose the reference the
             // writer can no longer reconstruct the contribution.
@@ -1905,10 +1970,10 @@ class QuicConnection(
             // peer's plausible retransmit horizon.
             recordRetiredStreamIdLocked(stream.streamId)
             streams.remove(stream.streamId)
-            it.remove()
             removed++
         }
         if (removed > 0) {
+            streamsList = kept
             retiredStreamsCount += removed.toLong()
             // The writer's round-robin cursor is a position in
             // [streamsList], so a removal that crossed the cursor would
@@ -2491,6 +2556,19 @@ class QuicConnection(
     }
 }
 
+/**
+ * Default monotonic clock supplier for [QuicConnection.nowMillis]. Returns a
+ * lambda that yields milliseconds-elapsed-since-construction, anchored on a
+ * fresh [kotlin.time.TimeSource.Monotonic.markNow] mark. Only differences are
+ * meaningful; an NTP step / suspend-resume does not affect the values.
+ */
+private fun defaultMonotonicNowMillis(): () -> Long {
+    val anchor =
+        kotlin.time.TimeSource.Monotonic
+            .markNow()
+    return { anchor.elapsedNow().inWholeMilliseconds }
+}
+
 /** Connection was closed (locally or by peer) before reaching CONNECTED. */
 class QuicConnectionClosedException(
     message: String,

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

@@ -52,8 +52,16 @@ class QuicConnectionDriver(
     val connection: QuicConnection,
     private val socket: UdpSocket,
     private val parentScope: CoroutineScope,
-    private val nowMillis: () -> Long = { System.currentTimeMillis() },
 ) {
+    /**
+     * Single time source: defer to the connection's [QuicConnection.nowMillis]
+     * (monotonic by default). Pre-fix the driver had its own
+     * `System.currentTimeMillis()` default which silently disagreed with the
+     * connection's wallclock-derived clock; under NTP step the two could
+     * report different values for the SAME logical "now", leading to RTT
+     * samples computed against drifted timestamps.
+     */
+    private val nowMillis: () -> Long get() = connection.nowMillis
     private val job = SupervisorJob(parentScope.coroutineContext[Job])
     private val scope = CoroutineScope(parentScope.coroutineContext + job + Dispatchers.IO)
     private val sendWakeup = Channel<Unit>(Channel.CONFLATED)
@@ -390,6 +398,13 @@ class QuicConnectionDriver(
  * (or the fresh one — both are valid) and is at worst a no-op.
  */
 internal suspend fun handlePtoFired(conn: QuicConnection) {
+    // Increment FIRST so [requeueInflightForProbe]'s threshold check
+    // sees the post-increment value. The increment must happen exactly
+    // once per PTO event — not per call to [requeueInflightForProbe],
+    // because the send loop calls that helper again between the first
+    // and second probe (RFC 9002 §6.2.4) and that re-requeue is part
+    // of the SAME PTO event.
+    conn.consecutivePtoCount = (conn.consecutivePtoCount + 1).coerceAtMost(6)
     conn.pendingPing = true
     requeueInflightForProbe(conn)
     // RFC 9002 §6.2.4: the spec allows up to 2 ack-eliciting packets
@@ -400,7 +415,6 @@ internal suspend fun handlePtoFired(conn: QuicConnection) {
     // nothing requeued yet" — the requeue is what makes the budget
     // meaningful.
     conn.pendingProbePackets = 2
-    conn.consecutivePtoCount = (conn.consecutivePtoCount + 1).coerceAtMost(6)
 }
 
 /**
@@ -426,12 +440,13 @@ internal suspend fun requeueInflightForProbe(conn: QuicConnection) {
     if (conn.initial.sendProtection != null && !conn.initial.keysDiscarded) {
         conn.requeueAllInflightCrypto(EncryptionLevel.INITIAL)
     }
-    // Bug-6 fix: increment BEFORE the threshold check. With the
-    // pre-fix ordering and threshold=2, the rotation actually fired
-    // on the 3rd PTO (count went 0→1→2 before check). Now the count
-    // matches the constant's natural reading: "after 2 consecutive
-    // PTOs with no progress, trigger migration on the 2nd PTO firing."
-    conn.consecutivePtoCount = (conn.consecutivePtoCount + 1).coerceAtMost(6)
+    // The PTO count is incremented in [handlePtoFired] BEFORE this
+    // helper runs, so the threshold check below sees the post-increment
+    // value (matching the constant's natural reading: "after N
+    // consecutive PTOs with no progress, trigger migration on the Nth
+    // PTO firing"). The send loop's between-probe re-requeue calls
+    // this helper without bumping the count again — that's the same
+    // PTO event, not a new one.
     // Once 1-RTT keys are installed, PTO must also retransmit application
     // data — STREAM bytes that were sent but never ACK'd. Without this,
     // a single corrupted/lost 1-RTT packet (especially the first one

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

@@ -71,6 +71,22 @@ fun feedDatagram(
     conn: QuicConnection,
     datagram: ByteArray,
     nowMillis: Long,
+) {
+    try {
+        feedDatagramInner(conn, datagram, nowMillis)
+    } catch (e: com.vitorpamplona.quic.QuicProtocolViolationException) {
+        // RFC 9000 §17.2 / §17.3.1: peer set reserved bits in the
+        // header after HP unmask (or a similar invariant violation
+        // bubbled out of the parse path). Spec MUST close with
+        // PROTOCOL_VIOLATION.
+        conn.markClosedExternally(e.message ?: "PROTOCOL_VIOLATION")
+    }
+}
+
+private fun feedDatagramInner(
+    conn: QuicConnection,
+    datagram: ByteArray,
+    nowMillis: Long,
 ) {
     var offset = 0
     while (offset < datagram.size) {
@@ -408,7 +424,16 @@ private fun feedShortHeaderPacket(
     // to [previousReceiveProtection], and rolls the send side forward so
     // the next outbound carries the matching KEY_PHASE bit (peer uses
     // that to confirm the rotation completed).
-    if (rotateOnSuccess != null) {
+    //
+    // RFC 9001 §6.1: a peer that initiated a key update MUST NOT send a
+    // post-rotation packet with a smaller PN than any pre-rotation
+    // packet. So a "next-phase" packet with PN <= largestReceived is
+    // either a misbehaving peer or an off-path attempt to flip our key
+    // state with a captured/forged packet. We refuse the commit in that
+    // case — AEAD already cleared, so the frames are still delivered,
+    // but we don't promote next-phase keys to current. Aligns with
+    // quicly / quiche / s2n-quic behaviour.
+    if (rotateOnSuccess != null && parsed.packet.packetNumber > state.pnSpace.largestReceived) {
         conn.commitKeyUpdate(rotateOnSuccess)
     }
     state.pnSpace.observeInbound(parsed.packet.packetNumber, nowMillis)
@@ -499,7 +524,19 @@ private fun dispatchFrames(
                     state.largestAckedSentTimeMs = largestSentTime
                 }
                 if (advancedLargest && largestSentTime != null && drained.any { it.ackEliciting }) {
-                    val ackDelayUs = frame.ackDelay shl conn.config.ackDelayExponent.toInt()
+                    // Peer-controlled `frame.ackDelay` is a varint up to 2^62-1.
+                    // Naive `shl ackDelayExponent` overflows Long for crafted
+                    // values (negative result → poisoned RTT sample inflating
+                    // smoothed_rtt). Clamp the exponent and the shift result
+                    // before consuming. Per RFC 9000 §18.2 the exponent is
+                    // 0..20; we further clamp ackDelay so the shift never
+                    // overflows (`ackDelay <= Long.MAX_VALUE >>> exponent`).
+                    val rawExponent = conn.config.ackDelayExponent
+                    val exponent = rawExponent.coerceIn(0L, 20L).toInt()
+                    val maxAckDelayPreShift =
+                        if (exponent == 0) Long.MAX_VALUE else Long.MAX_VALUE ushr exponent
+                    val safeAckDelay = frame.ackDelay.coerceIn(0L, maxAckDelayPreShift)
+                    val ackDelayUs = safeAckDelay shl exponent
                     val ackDelayMs = ackDelayUs / 1_000L
                     conn.lossDetection.onRttSample(
                         largestAckedSentTimeMs = largestSentTime,
@@ -602,7 +639,33 @@ private fun dispatchFrames(
                     )
                     return
                 }
-                stream.receive.insert(frame.offset, frame.data, frame.fin)
+                // RFC 9000 §4.5: enforce final-size invariants. The
+                // [com.vitorpamplona.quic.stream.ReceiveBuffer.insert]
+                // surface returns a typed result; map any non-OK result
+                // to a connection close with FINAL_SIZE_ERROR so the
+                // peer knows it just violated the spec instead of
+                // having its bytes silently dropped.
+                when (stream.receive.insert(frame.offset, frame.data, frame.fin)) {
+                    com.vitorpamplona.quic.stream.ReceiveBuffer.InsertResult.OK -> {
+                        Unit
+                    }
+
+                    com.vitorpamplona.quic.stream.ReceiveBuffer.InsertResult.OFFSET_PAST_FIN -> {
+                        conn.markClosedExternally(
+                            "FINAL_SIZE_ERROR: stream ${frame.streamId} frame ends at " +
+                                "${frame.offset + frame.data.size} past final size ${stream.receive.finOffset}",
+                        )
+                        return
+                    }
+
+                    com.vitorpamplona.quic.stream.ReceiveBuffer.InsertResult.FIN_CONFLICTS_WITH_PRIOR_FIN -> {
+                        conn.markClosedExternally(
+                            "FINAL_SIZE_ERROR: stream ${frame.streamId} second FIN at " +
+                                "${frame.offset + frame.data.size} disagrees with prior final size ${stream.receive.finOffset}",
+                        )
+                        return
+                    }
+                }
                 val data = stream.receive.readContiguous()
                 if (data.isNotEmpty()) {
                     // Round-4 perf #9: mark the stream as needing a flow-
@@ -698,17 +761,61 @@ private fun dispatchFrames(
                     )
                     return
                 }
+                // 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.
+                val target = conn.streamByIdLocked(frame.streamId)
+                if (target != null) {
+                    val priorFin = target.receive.finOffset
+                    val highestSeen = target.receive.highestObservedOffset()
+                    if (priorFin != null && frame.finalSize != priorFin) {
+                        conn.markClosedExternally(
+                            "FINAL_SIZE_ERROR: stream ${frame.streamId} RESET_STREAM finalSize " +
+                                "${frame.finalSize} disagrees with prior FIN size $priorFin",
+                        )
+                        return
+                    }
+                    if (frame.finalSize < highestSeen) {
+                        conn.markClosedExternally(
+                            "FINAL_SIZE_ERROR: stream ${frame.streamId} RESET_STREAM finalSize " +
+                                "${frame.finalSize} below already-observed offset $highestSeen",
+                        )
+                        return
+                    }
+                }
                 // Mark the peer's stream aborted and close our read side; the
                 // application sees a truncated incoming flow.
-                conn.streamByIdLocked(frame.streamId)?.closeIncoming()
+                target?.closeIncoming()
             }
 
             is StopSendingFrame -> {
-                // Round-4 #2: peer asks us to stop sending on its read side.
-                // We don't model an outbound abort yet — this is acknowledged
-                // and dropped. A future enhancement should emit RESET_STREAM
-                // back per RFC 9000 §3.5.
+                // RFC 9000 §3.5: peer asks us to stop sending on its read side.
+                // We MUST respond with RESET_STREAM carrying the same
+                // application error code so the peer can free its receive
+                // resources. Pre-fix this was silently dropped, so the peer
+                // would keep buffering bytes we kept emitting and the
+                // application kept paying CPU on a stream the peer no
+                // longer cared about.
+                //
+                // resetStream() is "first-call wins" so a duplicate
+                // STOP_SENDING (peer retransmit) doesn't redundantly mutate
+                // state. Only triggers for streams with an outgoing side —
+                // peer-uni (CLIENT_UNI from peer's perspective = SERVER_UNI
+                // here) has none, so the call is a defensive no-op.
                 ackEliciting = true
+                conn.streamByIdLocked(frame.streamId)?.let { stream ->
+                    val kind = StreamId.kindOf(frame.streamId)
+                    val hasLocalSend =
+                        kind == StreamId.Kind.CLIENT_BIDI ||
+                            kind == StreamId.Kind.SERVER_BIDI ||
+                            kind == StreamId.Kind.CLIENT_UNI
+                    if (hasLocalSend) {
+                        stream.resetStream(frame.applicationErrorCode)
+                    }
+                }
             }
 
             is NewTokenFrame -> {

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

@@ -762,8 +762,15 @@ private fun buildApplicationPacket(
                             fin = chunk.fin,
                         )
                     packetBudget -= chunk.data.size + 32
-                    connBudget -= chunk.data.size
-                    conn.sendConnectionFlowConsumed += chunk.data.size
+                    // RFC 9000 §4.1: connection-level flow credit caps
+                    // cumulative *new* bytes only — retransmits MUST
+                    // NOT debit further. Pre-fix every retransmit
+                    // chunk re-debited credit, eventually starving the
+                    // connection on lossy paths after a few PTO rounds.
+                    if (!chunk.isRetransmit) {
+                        connBudget -= chunk.data.size
+                        conn.sendConnectionFlowConsumed += chunk.data.size
+                    }
                 }
             }
             tierStart = tierEnd

quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/AckedPackets.kt

@@ -34,8 +34,14 @@ import com.vitorpamplona.quic.frame.AckFrame
  * from blowing up the iterator.
  *
  * The function does not allocate per-PN; it iterates by primitive
- * `Long` and invokes [block] for each ACK'd PN. Hot path on every
- * inbound ACK frame, so allocation matters.
+ * `Long` and invokes [block] for each ACK'd PN.
+ *
+ * NOTE: this primitive walks every PN in the range. Production code
+ * MUST use [drainAckedSentPackets] (which is bounded by the in-flight
+ * map) — a hostile peer with `firstAckRange = 2^62-1` would otherwise
+ * pin a core forever inside this loop. This entry point is retained
+ * for tests that intentionally inspect the on-wire range expansion
+ * with small, well-formed inputs.
  */
 inline fun forEachAckedPacketNumber(
     ack: AckFrame,
@@ -71,14 +77,62 @@ inline fun forEachAckedPacketNumber(
  * `quic/plans/2026-05-04-control-frame-retransmit.md`).
  *
  * Caller must hold the connection lock.
+ *
+ * DoS hardening: instead of walking every PN inside each ACK range
+ * (up to 2^62 with a hostile peer — single 8-byte varint pinning a
+ * core forever), we walk the [sentPackets] keys (bounded by the
+ * congestion window, typically ≤ a few thousand) and check
+ * membership against the parsed range list. O(N·M) where N = sent
+ * packets in flight, M = ACK ranges (bounded by packet payload
+ * size). A peer with `firstAckRange = 2^62-1` simply matches every
+ * legitimate in-flight PN once, then returns.
  */
 fun drainAckedSentPackets(
     sentPackets: MutableMap<Long, SentPacket>,
     ack: AckFrame,
 ): List<SentPacket> {
+    if (sentPackets.isEmpty()) return emptyList()
+    val ranges = parseAckRanges(ack)
+    if (ranges.isEmpty()) return emptyList()
     val drained = mutableListOf<SentPacket>()
-    forEachAckedPacketNumber(ack) { pn ->
-        sentPackets.remove(pn)?.let { drained += it }
+    val it = sentPackets.entries.iterator()
+    while (it.hasNext()) {
+        val entry = it.next()
+        val pn = entry.key
+        for (i in ranges.indices) {
+            val r = ranges[i]
+            if (pn in r) {
+                drained += entry.value
+                it.remove()
+                break
+            }
+        }
     }
     return drained
 }
+
+/**
+ * Parse an [AckFrame]'s on-wire ranges into a list of `[smallest, largest]`
+ * `LongRange` entries, descending. Each range is clamped to non-negative
+ * PNs and bounded against [Long] underflow on the additional-ranges
+ * walk. Stops early when the descending walk crosses zero (per RFC 9000
+ * §19.3.1, all PNs are non-negative). The result is small (bounded by
+ * the ACK frame's payload size).
+ */
+private fun parseAckRanges(ack: AckFrame): List<LongRange> {
+    val largest = ack.largestAcknowledged
+    if (largest < 0L) return emptyList()
+    val firstSmallest = (largest - ack.firstAckRange).coerceAtLeast(0L)
+    val out = ArrayList<LongRange>(ack.additionalRanges.size + 1)
+    out += firstSmallest..largest
+    var prevSmallest = firstSmallest
+    for (range in ack.additionalRanges) {
+        // RFC 9000 §19.3.1: nextLargest = prevSmallest - gap - 2.
+        val nextLargest = prevSmallest - range.gap - 2L
+        if (nextLargest < 0L) break
+        val nextSmallest = (nextLargest - range.ackRangeLength).coerceAtLeast(0L)
+        out += nextSmallest..nextLargest
+        prevSmallest = nextSmallest
+    }
+    return out
+}

quic/src/commonMain/kotlin/com/vitorpamplona/quic/crypto/InitialSecrets.kt

@@ -37,7 +37,14 @@ package com.vitorpamplona.quic.crypto
  * protection epoch.
  */
 object InitialSecrets {
-    val V1_INITIAL_SALT: ByteArray =
+    /**
+     * RFC 9001 §5.2 fixed Initial-secret salt for QUIC v1. Private + only
+     * read internally by [derive] — pre-fix the constant was a public
+     * mutable [ByteArray] that any caller could stomp on (or that any
+     * stack trace / `toString()` could leak). Crypto material doesn't
+     * need to be reachable outside the derive path.
+     */
+    private val V1_INITIAL_SALT: ByteArray =
         byteArrayOf(
             0x38.toByte(),
             0x76.toByte(),

quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3FrameReader.kt

@@ -33,13 +33,71 @@ import com.vitorpamplona.quic.Varint
  * Use one [Http3FrameReader] per HTTP/3 stream. Feed bytes via [push]; drain
  * frames via [next] until it returns null.
  *
- * Unknown frame types (per RFC 9114 §9 — "frame types in the format and intent
- * not recognized SHOULD be ignored") are surfaced as [Http3Frame.Unknown] so
- * the caller can decide policy. SETTINGS, HEADERS, DATA are typed.
+ * Stream-context enforcement (RFC 9114 §7.2):
+ *  * On the [StreamContext.CONTROL] stream, the FIRST frame MUST be SETTINGS
+ *    (else H3_MISSING_SETTINGS); DATA / HEADERS / PUSH_PROMISE are not
+ *    allowed (H3_FRAME_UNEXPECTED).
+ *  * On a [StreamContext.REQUEST] stream, SETTINGS / GOAWAY / MAX_PUSH_ID /
+ *    CANCEL_PUSH are not allowed (H3_FRAME_UNEXPECTED).
+ *  * On a [StreamContext.PUSH] stream, SETTINGS / GOAWAY / MAX_PUSH_ID /
+ *    PUSH_PROMISE / CANCEL_PUSH are not allowed.
+ *  * Reserved frame types `0x02, 0x06, 0x08, 0x09` are explicitly rejected
+ *    (H3_FRAME_UNEXPECTED) per RFC 9114 §7.2.8 — only the
+ *    `0x21 + 0x1f * N` reserved-greasing pattern is ignored as Unknown.
+ *  * On [StreamContext.WT_BIDI_DATA] / [StreamContext.WT_UNI_DATA]
+ *    (post-prefix WebTransport stream payload), no HTTP/3 framing applies
+ *    — the caller never feeds bytes through here. Provided for symmetry.
+ *
+ * Memory safety:
+ *  * Pending buffered bytes are capped at [maxPendingBytes]; a peer that
+ *    streams a partial-frame prefix without ever delivering the body
+ *    cannot pin unbounded heap. Exceeding the cap throws
+ *    [QuicCodecException], which the caller surfaces as a connection
+ *    error.
+ *  * Per-frame body length is capped by [maxFrameBodyBytes]. A peer that
+ *    advertises a 2 GiB length on the wire is rejected before the
+ *    `ByteArray(len.toInt())` allocation.
  */
-class Http3FrameReader {
+class Http3FrameReader(
+    private val context: StreamContext = StreamContext.UNCHECKED,
+    private val maxPendingBytes: Int = DEFAULT_MAX_PENDING_BYTES,
+    private val maxFrameBodyBytes: Int = DEFAULT_MAX_FRAME_BODY_BYTES,
+) {
     private var buf: ByteArray = ByteArray(0)
     private var pos: Int = 0
+    private var framesEmitted: Int = 0
+
+    /** Bytes currently buffered awaiting a complete frame. Diagnostic; tests use this. */
+    val bufferedBytes: Int get() = buf.size - pos
+
+    /**
+     * Stream context for [Http3FrameReader] frame validation. The reader is
+     * stateless about which kind of stream it's running on unless the
+     * caller tells it. [UNCHECKED] preserves pre-fix behaviour for tests
+     * that intentionally feed mixed frames; production callers should
+     * specify the actual context.
+     */
+    enum class StreamContext {
+        /** No per-context validation (pre-fix behaviour, tests). */
+        UNCHECKED,
+
+        /** RFC 9114 §6.2.1 server CONTROL unidi stream. */
+        CONTROL,
+
+        /** RFC 9114 §6.1 client/server REQUEST bidi stream. */
+        REQUEST,
+
+        /** RFC 9114 §6.2.2 PUSH unidi stream. */
+        PUSH,
+
+        /**
+         * Post-prefix WT bidi/uni stream payload. WebTransport bytes don't
+         * use HTTP/3 framing — this exists so the type system can prevent
+         * accidental use of [Http3FrameReader] on raw WT data.
+         */
+        WT_BIDI_DATA,
+        WT_UNI_DATA,
+    }
 
     fun push(bytes: ByteArray) {
         if (bytes.isEmpty()) return
@@ -50,6 +108,14 @@ class Http3FrameReader {
             buf = buf.copyOfRange(pos, buf.size)
             pos = 0
         }
+        val newPending = (buf.size - pos).toLong() + bytes.size.toLong()
+        if (newPending > maxPendingBytes) {
+            throw QuicCodecException(
+                "HTTP/3 frame reader buffer would exceed cap " +
+                    "($newPending > $maxPendingBytes); peer is streaming a partial " +
+                    "frame without delivering the body — likely H3_EXCESSIVE_LOAD",
+            )
+        }
         val combined = ByteArray(buf.size + bytes.size)
         buf.copyInto(combined, 0)
         bytes.copyInto(combined, buf.size)
@@ -63,21 +129,132 @@ class Http3FrameReader {
         val lenRes = Varint.decode(buf, typeEnd) ?: return null
         val bodyStart = typeEnd + lenRes.bytesConsumed
         val len = lenRes.value
-        if (len < 0 || len > Int.MAX_VALUE.toLong()) {
-            throw QuicCodecException("HTTP/3 frame length out of range: $len")
+        if (len < 0 || len > maxFrameBodyBytes.toLong()) {
+            throw QuicCodecException(
+                "HTTP/3 frame length out of range: $len (cap $maxFrameBodyBytes)",
+            )
         }
         val bodyEnd = bodyStart + len.toInt()
         if (bodyEnd > buf.size) return null // not all body bytes present yet
+        val type = typeRes.value
+        validateFrameType(type)
         val body = buf.copyOfRange(bodyStart, bodyEnd)
         pos = bodyEnd
-        return when (typeRes.value) {
+        framesEmitted++
+        return when (type) {
             Http3FrameType.DATA -> Http3Frame.Data(body)
             Http3FrameType.HEADERS -> Http3Frame.Headers(body)
             Http3FrameType.SETTINGS -> Http3Frame.Settings(Http3Settings.decodeBody(body))
             Http3FrameType.GOAWAY -> Http3Frame.Goaway(body)
-            else -> Http3Frame.Unknown(typeRes.value, body)
+            else -> Http3Frame.Unknown(type, body)
         }
     }
+
+    /**
+     * Enforce RFC 9114 §7.2 per-stream-context rules. Throws
+     * [QuicCodecException] (mapped upstream to H3_FRAME_UNEXPECTED /
+     * H3_MISSING_SETTINGS) on violation.
+     */
+    private fun validateFrameType(type: Long) {
+        // RFC 9114 §7.2.8: explicit reserved-and-forbidden frame types.
+        // Distinct from the reserved-greasing pattern `0x21 + 0x1f * N`
+        // (which is allowed and falls through as Unknown).
+        if (type == 0x02L || type == 0x06L || type == 0x08L || type == 0x09L) {
+            throw QuicCodecException(
+                "H3_FRAME_UNEXPECTED: reserved HTTP/3 frame type 0x${type.toString(16)}",
+            )
+        }
+        when (context) {
+            StreamContext.UNCHECKED -> {
+                Unit
+            }
+
+            StreamContext.CONTROL -> {
+                // §7.2.4: SETTINGS MUST be the first frame on the
+                // control stream; any non-SETTINGS first frame =>
+                // H3_MISSING_SETTINGS.
+                if (framesEmitted == 0 && type != Http3FrameType.SETTINGS) {
+                    throw QuicCodecException(
+                        "H3_MISSING_SETTINGS: first CONTROL-stream frame was 0x${type.toString(16)}",
+                    )
+                }
+                // §7.2.4: a second SETTINGS is also forbidden.
+                if (framesEmitted > 0 && type == Http3FrameType.SETTINGS) {
+                    throw QuicCodecException(
+                        "H3_FRAME_UNEXPECTED: duplicate SETTINGS on CONTROL stream",
+                    )
+                }
+                // §7.2.1, §7.2.2, §7.2.5: DATA / HEADERS / PUSH_PROMISE
+                // are forbidden on the control stream.
+                if (type == Http3FrameType.DATA ||
+                    type == Http3FrameType.HEADERS ||
+                    type == Http3FrameType.PUSH_PROMISE
+                ) {
+                    throw QuicCodecException(
+                        "H3_FRAME_UNEXPECTED: 0x${type.toString(16)} on CONTROL stream",
+                    )
+                }
+            }
+
+            StreamContext.REQUEST -> {
+                // §7.2.4 / §7.2.6 / §7.2.7: control-only frames are
+                // forbidden on request streams.
+                if (type == Http3FrameType.SETTINGS ||
+                    type == Http3FrameType.GOAWAY ||
+                    type == Http3FrameType.MAX_PUSH_ID ||
+                    type == Http3FrameType.CANCEL_PUSH
+                ) {
+                    throw QuicCodecException(
+                        "H3_FRAME_UNEXPECTED: 0x${type.toString(16)} on REQUEST stream",
+                    )
+                }
+            }
+
+            StreamContext.PUSH -> {
+                if (type == Http3FrameType.SETTINGS ||
+                    type == Http3FrameType.GOAWAY ||
+                    type == Http3FrameType.MAX_PUSH_ID ||
+                    type == Http3FrameType.PUSH_PROMISE ||
+                    type == Http3FrameType.CANCEL_PUSH
+                ) {
+                    throw QuicCodecException(
+                        "H3_FRAME_UNEXPECTED: 0x${type.toString(16)} on PUSH stream",
+                    )
+                }
+            }
+
+            StreamContext.WT_BIDI_DATA, StreamContext.WT_UNI_DATA -> {
+                // WebTransport stream payload doesn't use HTTP/3 framing.
+                // The caller shouldn't be feeding bytes through this
+                // reader for those contexts; flag any attempt loudly.
+                throw QuicCodecException(
+                    "WebTransport stream payload routed through Http3FrameReader",
+                )
+            }
+        }
+    }
+
+    companion object {
+        /**
+         * Default cap on pending unparsed bytes (1 MiB). This is the
+         * "frame in flight but body not yet complete" headroom — more
+         * than enough for a legitimate HEADERS / SETTINGS / WT capsule
+         * preamble plus a single in-progress DATA chunk, far less than
+         * the heap a hostile peer could pin.
+         */
+        const val DEFAULT_MAX_PENDING_BYTES: Int = 1 shl 20
+
+        /**
+         * Default cap on a single HTTP/3 frame body (16 MiB). Exceeds
+         * any plausible HEADERS payload (RFC 9114 only ever expects
+         * tens of KiB) and any legitimate per-frame DATA chunk that
+         * fits inside QUIC's per-packet payload budget. Setting this
+         * lower than [DEFAULT_MAX_PENDING_BYTES] would require
+         * splitting; setting it equal keeps the two caps aligned for
+         * single-frame stalls.
+         */
+        const val DEFAULT_MAX_FRAME_BODY_BYTES: Int = 16 shl 20
+    }
 }
 
 /** A parsed HTTP/3 frame. */

quic/src/commonMain/kotlin/com/vitorpamplona/quic/http3/Http3Settings.kt

@@ -63,10 +63,62 @@ data class Http3Settings(
                         "duplicate HTTP/3 SETTINGS id 0x${id.toString(16)}",
                     )
                 }
+                // RFC 9114 §7.2.4.1 / RFC 9204 §5: per-id range checks.
+                // A peer that advertises e.g. `MAX_FIELD_SECTION_SIZE =
+                // 2^60` could otherwise drive the encoder into
+                // unbounded heap (we'd emit headers up to that cap in a
+                // single allocation). Bounds chosen to comfortably
+                // exceed any legitimate value while staying inside
+                // [Int.MAX_VALUE] for safe `.toInt()` casts at use
+                // sites.
+                validateValue(id, value)
                 map[id] = value
             }
             return Http3Settings(map)
         }
+
+        /**
+         * Reject obviously-malicious SETTINGS values per the relevant
+         * RFCs. Negative varints are already impossible (varint is
+         * unsigned), but we double-check defensively.
+         */
+        private fun validateValue(
+            id: Long,
+            value: Long,
+        ) {
+            if (value < 0L) {
+                throw com.vitorpamplona.quic.QuicCodecException(
+                    "negative HTTP/3 SETTINGS value for id 0x${id.toString(16)}: $value",
+                )
+            }
+            // Per-id sanity caps. Unknown ids fall through (RFC 9114
+            // §7.2.4.1 says unknown SETTINGS MUST be ignored, but we
+            // still bound the value to avoid attacker-controlled
+            // long-tail allocation if any consumer ever uses the
+            // unknown id directly).
+            val cap: Long =
+                when (id) {
+                    Http3SettingsId.QPACK_MAX_TABLE_CAPACITY -> 1L shl 30
+
+                    // 1 GiB
+                    Http3SettingsId.MAX_FIELD_SECTION_SIZE -> 1L shl 30
+
+                    Http3SettingsId.QPACK_BLOCKED_STREAMS -> 65535L
+
+                    Http3SettingsId.ENABLE_CONNECT_PROTOCOL -> 1L
+
+                    Http3SettingsId.H3_DATAGRAM -> 1L
+
+                    Http3SettingsId.ENABLE_WEBTRANSPORT -> 1L
+
+                    else -> 1L shl 32 // generic cap for unknown ids
+                }
+            if (value > cap) {
+                throw com.vitorpamplona.quic.QuicCodecException(
+                    "HTTP/3 SETTINGS id 0x${id.toString(16)} value $value exceeds cap $cap",
+                )
+            }
+        }
     }
 }
 

quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/LongHeaderPacket.kt

@@ -183,10 +183,27 @@ object LongHeaderPacket {
         val packet = bytes.copyOfRange(packetStart, packetEnd)
         val localPnOffset = pnOffset - packetStart
 
-        // Step 1: unmask the first byte so we can read pnLen.
-        val firstByteMask = if ((first and 0x80) != 0) 0x0F else 0x1F
-        packet[0] = (first xor (mask[0].toInt() and firstByteMask)).toByte()
-        val pnLen = ((packet[0].toInt() and 0xFF) and 0x03) + 1
+        // Step 1: unmask the first byte so we can read pnLen. The
+        // long-header form bit (0x80) was already validated above (we
+        // wouldn't be in this function otherwise), so the first-byte
+        // mask is fixed at 0x0F — pre-fix the conditional `if (form == 1)`
+        // path was dead code.
+        packet[0] = (first xor (mask[0].toInt() and 0x0F)).toByte()
+        val unmaskedFirst = packet[0].toInt() and 0xFF
+        // RFC 9000 §17.2: long header layout is `1|1|T|T|R|R|P|P` —
+        // the two reserved bits at 0x0C MUST be zero after HP unmasking.
+        // A peer that sets either bit is in protocol violation. The
+        // form-bit (0x80) and fixed-bit (0x40) are not header-protected,
+        // so they're already correct; we skip the AEAD here on
+        // reserved-bit set rather than letting a malformed-but-AEAD-OK
+        // packet drift our state.
+        if ((unmaskedFirst and 0x0C) != 0) {
+            throw com.vitorpamplona.quic.QuicProtocolViolationException(
+                "PROTOCOL_VIOLATION: long-header reserved bits set " +
+                    "(0x${unmaskedFirst.toString(16)})",
+            )
+        }
+        val pnLen = (unmaskedFirst and 0x03) + 1
 
         // Step 2: unmask exactly `pnLen` packet-number bytes.
         for (i in 0 until pnLen) {

quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/RetryPacket.kt

@@ -107,8 +107,13 @@ data class RetryPacket(
             )
         }
 
-        /** RFC 9001 §5.8 — fixed retry-integrity AES-128-GCM key for QUIC v1. */
-        val V1_RETRY_KEY: ByteArray =
+        /**
+         * RFC 9001 §5.8 — fixed retry-integrity AES-128-GCM key for QUIC v1.
+         * Private — only read by [computeIntegrityTag] inside this file.
+         * Pre-fix exposing the public mutable [ByteArray] let any caller
+         * stomp on it or surface it via reflection / `toString()`.
+         */
+        private val V1_RETRY_KEY: ByteArray =
             byteArrayOf(
                 0xbe.toByte(),
                 0x0c.toByte(),
@@ -128,8 +133,11 @@ data class RetryPacket(
                 0x4e.toByte(),
             )
 
-        /** RFC 9001 §5.8 — fixed retry-integrity AES-128-GCM nonce for QUIC v1. */
-        val V1_RETRY_NONCE: ByteArray =
+        /**
+         * RFC 9001 §5.8 — fixed retry-integrity AES-128-GCM nonce for QUIC v1.
+         * Private; same rationale as [V1_RETRY_KEY].
+         */
+        private val V1_RETRY_NONCE: ByteArray =
             byteArrayOf(
                 0x46.toByte(),
                 0x15.toByte(),

quic/src/commonMain/kotlin/com/vitorpamplona/quic/packet/ShortHeaderPacket.kt

@@ -158,7 +158,20 @@ object ShortHeaderPacket {
         val localPnOffset = pnOffset - offset
         val firstByteMask = 0x1F
         packet[0] = (first xor (mask[0].toInt() and firstByteMask)).toByte()
-        val pnLen = ((packet[0].toInt() and 0xFF) and 0x03) + 1
+        val unmaskedFirst = packet[0].toInt() and 0xFF
+        // RFC 9000 §17.3.1: short header layout is `0|1|S|R|R|K|P|P` —
+        // the two reserved bits at 0x18 MUST be zero after HP unmasking.
+        // A peer that sets either bit is in protocol violation. Pre-fix
+        // we silently accepted the packet and let the AEAD pass; an
+        // off-spec server (or a fuzzer) could then push us to interop
+        // bugs that don't reproduce against well-behaved peers.
+        if ((unmaskedFirst and 0x18) != 0) {
+            throw com.vitorpamplona.quic.QuicProtocolViolationException(
+                "PROTOCOL_VIOLATION: short-header reserved bits set " +
+                    "(0x${unmaskedFirst.toString(16)})",
+            )
+        }
+        val pnLen = (unmaskedFirst and 0x03) + 1
         for (i in 0 until pnLen) {
             packet[localPnOffset + i] = (packet[localPnOffset + i].toInt() xor mask[1 + i].toInt()).toByte()
         }

quic/src/commonMain/kotlin/com/vitorpamplona/quic/qpack/QpackHuffman.kt

@@ -293,30 +293,54 @@ object QpackHuffman {
         )
 
     /**
-     * Lookup tables grouped by code length. `byLength[L]` is a HashMap from
-     * `code` (an Int up to 30 bits) to `symbol` (0..255) for all symbols
-     * whose Huffman code is exactly L bits long. Lengths used in the table
-     * range from 5 to 30. We omit the EOS (length 30, code 0x3FFFFFFF) since
-     * it must never appear in valid input.
+     * Lookup tables grouped by code length. For each length L in 5..30,
+     * [codesByLen]\[L\] holds the codes ascending and [symsByLen]\[L\]
+     * the matching symbol indices in lock-step — binary-search by the
+     * candidate Int gives the symbol with no boxing. Pre-fix this was
+     * `Array<HashMap<Int, Int>>` keyed by boxed `Integer`: every
+     * decoded character allocated a fresh `Integer` for the
+     * `candidate` plus a fresh `Integer` return wrapper, dominating
+     * GC pressure on every QPACK header decode. We omit the EOS
+     * (length 30, code 0x3FFFFFFF) since it must never appear in
+     * valid input.
      */
-    private val byLength: Array<HashMap<Int, Int>> = buildLookupByLength()
+    private val codesByLen: Array<IntArray>
+    private val symsByLen: Array<IntArray>
 
     /** Sorted ascending list of distinct code lengths actually used by the table. */
-    private val lengths: IntArray = byLength.indices.filter { byLength[it].isNotEmpty() }.toIntArray()
+    private val lengths: IntArray
 
-    private fun buildLookupByLength(): Array<HashMap<Int, Int>> {
-        val out = Array(31) { HashMap<Int, Int>() }
-        for (sym in 0..255) {
-            val code = table[sym][0]
-            val len = table[sym][1]
-            out[len][code] = sym
+    init {
+        // Allocate slots for lengths up to and including 30 (RFC 7541's
+        // longest non-EOS code is 30 bits — symbols 10/13/22 — plus EOS
+        // itself which we exclude by iterating 0..255 below).
+        val codes = Array(31) { IntArray(0) }
+        val syms = Array(31) { IntArray(0) }
+        for (len in 5..30) {
+            // Collect all symbols whose code length equals `len`, sorted
+            // by code so we can binary-search at decode time. Iterating
+            // 0..255 (not 0..256) silently excludes EOS, the only
+            // length-30 entry that must never appear in valid input.
+            val pairs = (0..255).filter { table[it][1] == len }.map { it to table[it][0] }
+            val sorted = pairs.sortedBy { it.second }
+            codes[len] = IntArray(sorted.size) { sorted[it].second }
+            syms[len] = IntArray(sorted.size) { sorted[it].first }
         }
-        return out
+        codesByLen = codes
+        symsByLen = syms
+        lengths = (5..30).filter { codesByLen[it].isNotEmpty() }.toIntArray()
     }
 
     /** Decode a Huffman-encoded byte sequence into a UTF-8 string. */
     fun decode(encoded: ByteArray): ByteArray {
-        val result = ArrayList<Byte>(encoded.size * 2) // rough upper bound
+        // Output is a growable ByteArray rather than ArrayList<Byte>:
+        // the latter boxes every emitted byte through `java.lang.Byte`,
+        // which on a 64-bit JVM is ~16 bytes of wrapper per output byte.
+        // Pre-grow to ~2× input as a rough upper bound; ASCII-heavy
+        // headers compress to ~62% with HPACK Huffman, so 2× input is
+        // an overestimate and we rarely need to grow.
+        var out = ByteArray(maxOf(8, encoded.size * 2))
+        var outPos = 0
         var bitBuf = 0L
         var bitsAvailable = 0
         var i = 0
@@ -333,9 +357,12 @@ object QpackHuffman {
             for (len in lengths) {
                 if (len > bitsAvailable) break
                 val candidate = ((bitBuf ushr (bitsAvailable - len)) and ((1L shl len) - 1)).toInt()
-                val sym = byLength[len][candidate]
-                if (sym != null) {
-                    result.add(sym.toByte())
+                val codes = codesByLen[len]
+                val pos = codes.binarySearch(candidate)
+                if (pos >= 0) {
+                    val sym = symsByLen[len][pos]
+                    if (outPos == out.size) out = out.copyOf(out.size * 2)
+                    out[outPos++] = sym.toByte()
                     bitsAvailable -= len
                     bitBuf = bitBuf and ((1L shl bitsAvailable) - 1)
                     matched = true
@@ -354,6 +381,6 @@ object QpackHuffman {
                 throw QuicCodecException("invalid Huffman bit stream")
             }
         }
-        return result.toByteArray()
+        return if (outPos == out.size) out else out.copyOf(outPos)
     }
 }

quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt

@@ -55,6 +55,7 @@ class AckTracker {
             val r = ranges[i]
             if (packetNumber > r.endInclusive + 1) {
                 ranges.add(i, LongRange(packetNumber, packetNumber))
+                trimToMaxRanges()
                 return
             }
             if (packetNumber == r.endInclusive + 1) {
@@ -80,6 +81,25 @@ class AckTracker {
             }
         }
         ranges += LongRange(packetNumber, packetNumber)
+        trimToMaxRanges()
+    }
+
+    /**
+     * Cap [ranges] at [MAX_STORED_RANGES] by dropping the oldest entries
+     * (the smallest PNs, at the tail of the descending list). Without
+     * this a peer that sends a continuous stream of sparse PNs (every
+     * other one, alternating-bit pattern) inflates the range list
+     * unboundedly until the encoded ACK frame no longer fits in a
+     * packet. The trade-off is that the very oldest ranges become
+     * implicitly "unknown to us"; if the peer retransmits a packet from
+     * an evicted range we'll re-deliver it and the deduplication is
+     * downstream's problem (per-stream offset bookkeeping). Spurious
+     * retransmits on truly ancient PNs are negligible in practice.
+     */
+    private fun trimToMaxRanges() {
+        while (ranges.size > MAX_STORED_RANGES) {
+            ranges.removeAt(ranges.lastIndex)
+        }
     }
 
     fun hasUnackedAckEliciting(): Boolean = ackElicitingPending
@@ -143,8 +163,15 @@ class AckTracker {
             val len = ranges[i].endInclusive - ranges[i].start
             rest += AckRange(gap, len)
         }
-        val ackDelayMicros = (nowMillis - largestRecvTimeMillis) * 1000L
-        val ackDelay = (ackDelayMicros ushr ackDelayExponent).coerceAtLeast(0L)
+        // Clamp ≥ 0 BEFORE the shift, not after: `ushr` on a negative
+        // Long produces a giant positive value that the peer's RTT
+        // estimator would interpret as a multi-hour delay, poisoning
+        // its smoothed_rtt below min_rtt. Clock can move backwards if
+        // [nowMillis] is wallclock-derived (NTP step) — even though we
+        // default to a monotonic source, the ctor allows a custom
+        // supplier and tests inject virtual clocks.
+        val rawDelayMicros = (nowMillis - largestRecvTimeMillis).coerceAtLeast(0L) * 1000L
+        val ackDelay = rawDelayMicros ushr ackDelayExponent
         ackElicitingPending = false
         return AckFrame(
             largestAcknowledged = largest,
@@ -153,4 +180,14 @@ class AckTracker {
             additionalRanges = rest,
         )
     }
+
+    companion object {
+        /**
+         * Cap on stored disjoint ranges. 64 is far above any healthy-link
+         * count (a single in-order receive needs 1 range; tens of
+         * reorders per second push that to 4–8). Above this we evict
+         * oldest first.
+         */
+        const val MAX_STORED_RANGES: Int = 64
+    }
 }

quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt

@@ -220,13 +220,26 @@ class QuicStream(
      * (two app threads racing the writer's clear-after-emit).
      */
     fun resetStream(errorCode: Long) {
-        if (resetState != null) return
-        resetState =
-            ResetState(
-                errorCode = errorCode,
-                finalSize = send.nextOffset,
-            )
-        resetEmitPending = true
+        // Synchronized atomic compare-and-set: pre-fix the
+        // `if (resetState != null) return` plus the assignment was
+        // racy. Two concurrent callers (e.g. the application aborting
+        // a request while STOP_SENDING from the peer triggers our own
+        // resetStream from the parser) could both observe null and
+        // both write — the second write would clobber the first
+        // errorCode while [resetEmitPending] was already set. The
+        // writer would then emit a RESET_STREAM with whichever
+        // errorCode landed last, possibly different from what the
+        // application asked for. The synchronized block makes
+        // first-call-wins genuinely first-call-wins.
+        synchronized(this) {
+            if (resetState != null) return
+            resetState =
+                ResetState(
+                    errorCode = errorCode,
+                    finalSize = send.nextOffset,
+                )
+            resetEmitPending = true
+        }
     }
 
     /**
@@ -244,9 +257,12 @@ class QuicStream(
      * original frame already on the wire).
      */
     fun stopSending(errorCode: Long) {
-        if (stopSendingState != null) return
-        stopSendingState = StopSendingState(errorCode = errorCode)
-        stopSendingEmitPending = true
+        // Same atomic-CAS rationale as [resetStream].
+        synchronized(this) {
+            if (stopSendingState != null) return
+            stopSendingState = StopSendingState(errorCode = errorCode)
+            stopSendingEmitPending = true
+        }
     }
 
     /**

quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/ReceiveBuffer.kt

@@ -25,19 +25,31 @@ package com.vitorpamplona.quic.stream
  * (or for the per-encryption-level CRYPTO offset stream).
  *
  * Chunks may arrive in any order with possibly overlapping ranges. The buffer
- * coalesces them into a single contiguous prefix that's available to the
- * consumer via [readContiguous]. We never expose bytes past the contiguous
- * frontier — gaps stall further reads until the missing offsets arrive.
+ * stores them as **non-overlapping, sorted-by-offset segments** without
+ * eager coalescing — the contiguous prefix is collected at [readContiguous]
+ * time. Pre-fix the buffer coalesced overlapping segments on EVERY [insert]
+ * by allocating a fresh merged `ByteArray((hi - lo).toInt())` and copying
+ * each existing segment into it; under a 200-chunk reorder burst that's
+ * O(N²) bytes copied. The new layout is O(1) amortized per [insert] for
+ * the common cases (adjacent or non-overlapping) and only allocates at
+ * read time, when the consumer is ready to pull the bytes.
  *
- * For a fully streamed CRYPTO transcript, the consumer just reads contiguous
- * bytes whenever new data arrives; the lookup is O(log N) on the gap tree.
+ * Invariants on [chunks]:
+ *  * Sorted strictly ascending by `offset`.
+ *  * No two segments overlap, but adjacent segments (one ending at the
+ *    next's start) are allowed.
+ *  * No segment lives below [readOffset].
  *
- * Implementation: we store raw chunks sorted by offset. Calls to [insert]
- * coalesce adjacent and overlapping ranges. [readContiguous] returns the
- * range from the current cursor up to the first gap.
+ * For a fully streamed CRYPTO transcript the consumer just reads contiguous
+ * bytes whenever new data arrives.
  */
 class ReceiveBuffer {
-    private val chunks = mutableListOf<Chunk>() // sorted by offset, non-overlapping after insert
+    /** Sorted ascending by [Chunk.offset]; non-overlapping. See class kdoc. */
+    private val chunks = ArrayDeque<Chunk>()
+
+    /** Cached `chunks.sumOf { data.size }` so [bufferedAhead] is O(1). */
+    private var bufferedAheadBytes: Long = 0L
+
     var readOffset: Long = 0L
         private set
 
@@ -54,27 +66,42 @@ class ReceiveBuffer {
     var finOffset: Long? = null
         private set
 
-    /** Insert a chunk at [offset] of size [data.size]. Idempotent on overlap. */
+    /**
+     * Insert a chunk at [offset] of size [data.size]. Idempotent on overlap.
+     *
+     * Returns [InsertResult.OK] for a well-formed frame, or one of the
+     * RFC 9000 §4.5 final-size error variants when the peer's frame is
+     * inconsistent with state we've already accepted. The caller (the
+     * QUIC parser) is expected to translate these into a connection
+     * close with `FINAL_SIZE_ERROR`. Pre-fix the buffer silently
+     * dropped conflicting FINs and accepted post-FIN extension data,
+     * letting a buggy peer drift state without termination.
+     */
     fun insert(
         offset: Long,
         data: ByteArray,
         fin: Boolean = false,
-    ) {
-        if (data.isEmpty() && !fin) return
+    ): InsertResult {
+        if (data.isEmpty() && !fin) return InsertResult.OK
+        val frameEnd = offset + data.size
+        // RFC 9000 §4.5: once a final size is established, no STREAM
+        // frame may extend past it, and a second FIN must agree.
+        finOffset?.let { existing ->
+            if (fin && frameEnd != existing) {
+                return InsertResult.FIN_CONFLICTS_WITH_PRIOR_FIN
+            }
+            if (frameEnd > existing) {
+                return InsertResult.OFFSET_PAST_FIN
+            }
+        }
         if (fin) {
             finReceived = true
-            // The FIN flag carries an implicit final offset = offset + data.size.
-            // RFC 9000 §4.5: once set, this MUST NOT change; ignore subsequent
-            // FIN frames whose final size disagrees (they should already have
-            // been rejected at the stream-state level, but be defensive here).
-            val finalSize = offset + data.size
-            if (finOffset == null) finOffset = finalSize
+            if (finOffset == null) finOffset = frameEnd
         }
-        if (data.isEmpty()) return
+        if (data.isEmpty()) return InsertResult.OK
 
-        val end = offset + data.size
-        // Drop chunk parts already consumed.
-        if (end <= readOffset) return
+        // Drop chunk parts already consumed by the reader.
+        if (frameEnd <= readOffset) return InsertResult.OK
         val effOffset: Long
         val effData: ByteArray
         if (offset < readOffset) {
@@ -86,54 +113,170 @@ class ReceiveBuffer {
             effData = data
         }
 
-        // Find the first chunk that's not strictly before the new range. The
-        // boundary is `<=` so a perfectly adjacent prior chunk (its endOffset
-        // equals our offset) is included in the merge — otherwise it would
-        // stay as a separate adjacent chunk and bufferedAhead() would
-        // overcount on perfectly-sequential receives starting at offset > 0.
-        var startIdx = 0
-        while (startIdx < chunks.size && chunks[startIdx].endOffset() < effOffset) startIdx++
-        var endIdx = startIdx
-        while (endIdx < chunks.size && chunks[endIdx].offset <= effOffset + effData.size) endIdx++
-        // Also pull in the prior chunk if it's exactly adjacent on the lower end.
-        if (startIdx > 0 && chunks[startIdx - 1].endOffset() == effOffset) startIdx -= 1
-
-        if (startIdx == endIdx) {
-            // No overlap — just insert.
-            chunks.add(startIdx, Chunk(effOffset, effData))
-            return
-        }
-
-        // Coalesce [startIdx, endIdx) plus the new chunk.
-        var lo = effOffset
-        var hi = effOffset + effData.size
-        for (i in startIdx until endIdx) {
-            lo = minOf(lo, chunks[i].offset)
-            hi = maxOf(hi, chunks[i].endOffset())
-        }
-        val merged = ByteArray((hi - lo).toInt())
-        for (i in startIdx until endIdx) {
-            chunks[i].data.copyInto(merged, (chunks[i].offset - lo).toInt())
-        }
-        effData.copyInto(merged, (effOffset - lo).toInt())
-        // Replace
-        for (i in 1..(endIdx - startIdx)) chunks.removeAt(startIdx)
-        chunks.add(startIdx, Chunk(lo, merged))
+        insertNonOverlapping(effOffset, effData)
+        return InsertResult.OK
     }
 
-    /** Returns and consumes the contiguous bytes available starting from [readOffset]. */
+    /**
+     * Insert [data] starting at [start] into [chunks], trimming portions
+     * that already exist (retransmit / overlap). The result preserves the
+     * non-overlapping, sorted invariants. No allocation when the new
+     * range is fully covered (retransmit no-op) or when it doesn't
+     * touch any existing segment (single insert).
+     */
+    private fun insertNonOverlapping(
+        start: Long,
+        data: ByteArray,
+    ) {
+        val end = start + data.size
+        // Locate the first existing chunk whose `endOffset > start` —
+        // anything before it is strictly to the left of [start, end).
+        var idx = lowerBoundEnd(start)
+        var cursor = start
+
+        while (cursor < end) {
+            val existing = if (idx < chunks.size) chunks[idx] else null
+            val existingStart = existing?.offset ?: Long.MAX_VALUE
+            if (existingStart >= end) {
+                // Tail (or whole new range) sits past every existing
+                // segment that could overlap. Emit the remainder.
+                emitSegment(start, data, cursor, end)
+                cursor = end
+            } else if (existingStart > cursor) {
+                // Gap from [cursor, existingStart) — emit it, then skip
+                // over the existing segment (it covers some of [start,end)).
+                emitSegment(start, data, cursor, existingStart)
+                cursor = minOf(existing!!.endOffset, end)
+                idx++
+            } else {
+                // existingStart <= cursor — the existing segment covers
+                // [cursor, existing.endOffset). Skip ahead.
+                cursor = minOf(existing!!.endOffset, end)
+                idx++
+            }
+        }
+    }
+
+    /**
+     * Emit the slice of [src] covering absolute range `[segStart, segEnd)`
+     * as a new chunk, inserted in sorted order. [src] starts at absolute
+     * offset [srcStart], so the slice index is `(segStart - srcStart)`.
+     */
+    private fun emitSegment(
+        srcStart: Long,
+        src: ByteArray,
+        segStart: Long,
+        segEnd: Long,
+    ) {
+        if (segStart >= segEnd) return
+        val from = (segStart - srcStart).toInt()
+        val to = (segEnd - srcStart).toInt()
+        val piece =
+            if (from == 0 && to == src.size) {
+                // Whole src is the segment; avoid the copyOfRange.
+                src
+            } else {
+                src.copyOfRange(from, to)
+            }
+        // Insert in sorted position. The index from [lowerBoundOffset]
+        // is monotonically non-decreasing across the walk in
+        // [insertNonOverlapping] but we recompute defensively to keep
+        // this helper self-contained.
+        val insertAt = lowerBoundOffset(segStart)
+        chunks.add(insertAt, Chunk(segStart, piece))
+        bufferedAheadBytes += piece.size
+    }
+
+    /** Index of first chunk whose `endOffset > target`, or `chunks.size`. */
+    private fun lowerBoundEnd(target: Long): Int {
+        var lo = 0
+        var hi = chunks.size
+        while (lo < hi) {
+            val mid = (lo + hi) ushr 1
+            if (chunks[mid].endOffset > target) hi = mid else lo = mid + 1
+        }
+        return lo
+    }
+
+    /** Index of first chunk whose `offset >= target`, or `chunks.size`. */
+    private fun lowerBoundOffset(target: Long): Int {
+        var lo = 0
+        var hi = chunks.size
+        while (lo < hi) {
+            val mid = (lo + hi) ushr 1
+            if (chunks[mid].offset >= target) hi = mid else lo = mid + 1
+        }
+        return lo
+    }
+
+    /** Highest `offset + data.size` ever seen on this stream (whether or not contiguous). */
+    fun highestObservedOffset(): Long {
+        val finEnd = finOffset
+        val bufEnd = if (chunks.isEmpty()) readOffset else chunks.last().endOffset
+        return if (finEnd != null) maxOf(finEnd, bufEnd) else bufEnd
+    }
+
+    /** Result of an [insert] call — see RFC 9000 §4.5. */
+    enum class InsertResult {
+        /** Frame accepted (or harmlessly redundant). */
+        OK,
+
+        /**
+         * The frame's `offset + data.size` exceeds the previously-established
+         * final size. Caller MUST close with FINAL_SIZE_ERROR.
+         */
+        OFFSET_PAST_FIN,
+
+        /**
+         * A second FIN-bearing frame disagreed with the established final
+         * size. Caller MUST close with FINAL_SIZE_ERROR.
+         */
+        FIN_CONFLICTS_WITH_PRIOR_FIN,
+    }
+
+    /**
+     * Returns and consumes the contiguous bytes available starting from
+     * [readOffset]. Walks every consecutive segment whose left edge meets
+     * the current read frontier and returns them as a single concatenated
+     * [ByteArray]. Returns an empty array if the next pending segment is
+     * beyond the current frontier (gap) or the buffer is empty.
+     *
+     * Allocation: zero allocations when only one segment is consecutive
+     * (the most common case — one inserted segment per call); a single
+     * concat allocation when multiple segments stack up after a gap fill.
+     */
     fun readContiguous(): ByteArray {
-        if (chunks.isEmpty()) return ByteArray(0)
-        val first = chunks[0]
-        if (first.offset != readOffset) return ByteArray(0)
-        val data = first.data
-        readOffset += data.size
-        chunks.removeAt(0)
-        return data
+        if (chunks.isEmpty() || chunks.first().offset != readOffset) return EMPTY
+        val first = chunks.removeFirst()
+        bufferedAheadBytes -= first.data.size
+        var cursor = first.endOffset
+        // Fast-path: only one consecutive segment.
+        if (chunks.isEmpty() || chunks.first().offset != cursor) {
+            readOffset = cursor
+            return first.data
+        }
+        // Multiple consecutive segments — collect, then concat once.
+        val collected = mutableListOf(first)
+        var total = first.data.size.toLong()
+        while (chunks.isNotEmpty() && chunks.first().offset == cursor) {
+            val next = chunks.removeFirst()
+            bufferedAheadBytes -= next.data.size
+            cursor = next.endOffset
+            collected += next
+            total += next.data.size
+        }
+        val out = ByteArray(total.toInt())
+        var pos = 0
+        for (c in collected) {
+            c.data.copyInto(out, pos)
+            pos += c.data.size
+        }
+        readOffset = cursor
+        return out
     }
 
     /** Bytes already buffered and held back due to gaps. */
-    fun bufferedAhead(): Long = chunks.sumOf { it.data.size.toLong() }
+    fun bufferedAhead(): Long = bufferedAheadBytes
 
     /** Highest contiguous offset received so far. */
     fun contiguousEnd(): Long = readOffset
@@ -150,6 +293,10 @@ class ReceiveBuffer {
         val offset: Long,
         val data: ByteArray,
     ) {
-        fun endOffset() = offset + data.size
+        val endOffset: Long get() = offset + data.size
+    }
+
+    private companion object {
+        private val EMPTY = ByteArray(0)
     }
 }

quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt

@@ -223,7 +223,7 @@ class SendBuffer(
                 }
                 addToInFlight(Range(retransmitHead.offset, take, fin))
                 if (fin) _finSent = true
-                return@synchronized Chunk(retransmitHead.offset, payload, fin)
+                return@synchronized Chunk(retransmitHead.offset, payload, fin, isRetransmit = true)
             }
 
             // 2. Fresh bytes.
@@ -237,14 +237,14 @@ class SendBuffer(
                 val finForThis = _finPending && !_finSent && nextSendOffset == _nextOffset
                 addToInFlight(Range(offset, take, finForThis))
                 if (finForThis) _finSent = true
-                return@synchronized Chunk(offset, payload, finForThis)
+                return@synchronized Chunk(offset, payload, finForThis, isRetransmit = false)
             }
 
             // 3. FIN-only.
             if (_finPending && !_finSent) {
                 _finSent = true
                 addToInFlight(Range(nextSendOffset, 0L, true))
-                return@synchronized Chunk(nextSendOffset, ByteArray(0), true)
+                return@synchronized Chunk(nextSendOffset, ByteArray(0), true, isRetransmit = false)
             }
             null
         }
@@ -591,6 +591,23 @@ class SendBuffer(
         }
         dataLen -= advanceInt
         flushedFloor += advance
+        // Shrink the backing buffer when a transient burst has been fully
+        // drained. Without this, a stream that ever held N bytes pins
+        // `data.size = N` for the rest of the connection — long-tail
+        // memory retention. Trigger when:
+        //   * the buffer is large enough to bother (above the doubling
+        //     floor of 64 bytes) AND
+        //   * live data fits in 1/4 of the allocation (capacity is
+        //     ≥ 4× live size).
+        // Shrink to twice the live size (or [SHRINK_FLOOR_BYTES],
+        // whichever is larger) to leave headroom for the next push
+        // without re-doubling immediately.
+        if (data.size > SHRINK_FLOOR_BYTES && dataLen * 4 < data.size) {
+            val newCap = maxOf(SHRINK_FLOOR_BYTES, dataLen * 2)
+            if (newCap < data.size) {
+                data = data.copyOf(newCap)
+            }
+        }
     }
 
     /**
@@ -614,6 +631,17 @@ class SendBuffer(
         }
     }
 
+    private companion object {
+        /**
+         * Backing-buffer floor below which [advanceFlushedFloorIfPossible]
+         * does NOT shrink. Below this size the doubling-on-grow is so
+         * cheap that shrinking would be a wash; above it, transient
+         * bursts that bloat the buffer to MiBs leave a long-tail
+         * memory footprint we want to release.
+         */
+        const val SHRINK_FLOOR_BYTES: Int = 4096
+    }
+
     /**
      * One contiguous offset range tracked by the buffer's bookkeeping.
      * [length] is `Long` to match QUIC's offset arithmetic — practical
@@ -625,21 +653,36 @@ class SendBuffer(
         val fin: Boolean,
     )
 
+    /**
+     * One emit-able STREAM chunk. [isRetransmit] distinguishes a chunk
+     * pulled off the retransmit queue (path 1 of [takeChunk]) from a
+     * fresh-bytes emission (path 2/3). Connection-level flow control
+     * (RFC 9000 §4.1) caps cumulative *new* bytes — retransmits MUST
+     * NOT consume additional credit, so the writer reads this flag and
+     * skips `sendConnectionFlowConsumed += data.size` when it's true.
+     * Without the distinction a single round of loss recovery on a
+     * long stream eventually exhausts credit and stalls the connection.
+     */
     data class Chunk(
         val offset: Long,
         val data: ByteArray,
         val fin: Boolean,
+        val isRetransmit: Boolean = false,
     ) {
         // ByteArray needs explicit equality.
         override fun equals(other: Any?): Boolean {
             if (this === other) return true
             if (other !is Chunk) return false
-            return offset == other.offset && fin == other.fin && data.contentEquals(other.data)
+            return offset == other.offset &&
+                fin == other.fin &&
+                isRetransmit == other.isRetransmit &&
+                data.contentEquals(other.data)
         }
 
         override fun hashCode(): Int {
             var result = offset.hashCode()
             result = 31 * result + fin.hashCode()
+            result = 31 * result + isRetransmit.hashCode()
             result = 31 * result + data.contentHashCode()
             return result
         }

quic/src/commonMain/kotlin/com/vitorpamplona/quic/tls/TlsHandshakeMessages.kt

@@ -77,6 +77,13 @@ data class TlsServerHello(
             val cipherSuite = r.readUint16()
             r.readByte() // legacy_compression_method = 0
             val extensions = TlsExtension.decodeList(r)
+            // RFC 8446 §4.1.3: ServerHello body ends with the extensions
+            // block. Any trailing bytes are a malformed handshake message.
+            if (r.remaining != 0) {
+                throw QuicCodecException(
+                    "ServerHello has ${r.remaining} trailing bytes after extensions",
+                )
+            }
             return TlsServerHello(random, sessionId, cipherSuite, extensions)
         }
     }
@@ -92,14 +99,38 @@ data class TlsEncryptedExtensions(
     val alpn: ByteArray?
         get() =
             extensions.firstOrNull { it.type == TlsConstants.EXT_ALPN }?.data?.let {
-                // ALPN response carries a single protocol_name<1..2^8-1> inside protocols<3..2^16-1>
+                // ALPN response carries EXACTLY one protocol_name<1..2^8-1>
+                // inside protocols<3..2^16-1> per RFC 7301 §3.1. Reject
+                // multi-name responses — a server that returns more than
+                // one is in protocol violation, and silently picking the
+                // first masks an interop bug.
                 val r = QuicReader(it)
-                r.skip(2) // outer length
-                r.readTlsOpaque1()
+                val outerLen = r.readUint16()
+                if (outerLen != r.remaining) {
+                    throw QuicCodecException(
+                        "ALPN extension outerLen=$outerLen does not match remaining ${r.remaining}",
+                    )
+                }
+                val name = r.readTlsOpaque1()
+                if (r.remaining != 0) {
+                    throw QuicCodecException(
+                        "ALPN response carries multiple protocol names (RFC 7301 §3.1 forbids)",
+                    )
+                }
+                name
             }
 
     companion object {
-        fun decodeBody(r: QuicReader): TlsEncryptedExtensions = TlsEncryptedExtensions(TlsExtension.decodeList(r))
+        fun decodeBody(r: QuicReader): TlsEncryptedExtensions {
+            val extensions = TlsExtension.decodeList(r)
+            // RFC 8446 §4.3.1: EE body is exactly the extensions block.
+            if (r.remaining != 0) {
+                throw QuicCodecException(
+                    "EncryptedExtensions has ${r.remaining} trailing bytes",
+                )
+            }
+            return TlsEncryptedExtensions(extensions)
+        }
     }
 }
 
@@ -115,6 +146,17 @@ data class TlsCertificateChain(
         fun decodeBody(r: QuicReader): TlsCertificateChain {
             val ctx = r.readTlsOpaque1()
             val listLen = r.readUint24()
+            // Pre-fix: a malformed peer setting `listLen` larger than the
+            // remaining body bytes caused `end = r.position + listLen` to
+            // point past `r.limit`. The while-loop then ran on garbage
+            // until each `readTlsOpaque*` ran off the end and threw —
+            // delivering corrupt cert blobs to certs[] before that
+            // happened. Bound `end` against `r.limit` up front.
+            if (listLen > r.remaining) {
+                throw QuicCodecException(
+                    "Certificate listLen=$listLen exceeds remaining body ${r.remaining}",
+                )
+            }
             val end = r.position + listLen
             val certs = mutableListOf<ByteArray>()
             while (r.position < end) {
@@ -123,6 +165,14 @@ data class TlsCertificateChain(
                 r.readTlsOpaque2()
                 certs += cert
             }
+            // RFC 8446 §4.4.2: the certificate_list ends at exactly `end`;
+            // if a per-cert extensions length pushed us past it, the cert
+            // chain is malformed and we MUST close.
+            if (r.position != end) {
+                throw QuicCodecException(
+                    "Certificate listLen mismatch: position=${r.position} expected=$end",
+                )
+            }
             return TlsCertificateChain(ctx, certs)
         }
     }

quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt

@@ -223,6 +223,13 @@ class QuicWebTransportSessionState(
         connection.streamById(connectStreamId)?.let {
             it.send.enqueue(encodeCloseSessionCapsule(errorCode, reason))
             it.send.finish()
+            // Wake the driver BEFORE asking it to close — otherwise
+            // [driver.close] may short-circuit the send loop before our
+            // newly-enqueued WT_CLOSE_SESSION capsule + FIN drain to
+            // the wire. The peer would then see an abrupt UDP-level
+            // tear-down instead of the graceful WT close, and any
+            // application-error-code we wanted to deliver is lost.
+            driver.wakeup()
         }
         driver.close()
         // Round-5 concurrency #1: cancel the WT scope so the demux pump

quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt

@@ -48,12 +48,22 @@ fun encodeCloseSessionCapsule(
     errorCode: Int = 0,
     reason: String = "",
 ): ByteArray {
+    val reasonBytes = reason.encodeToByteArray()
+    // Symmetric with the decoder: draft-ietf-webtrans-http3 §5 caps the
+    // reason string at 8192 bytes. Catching this on encode means a local
+    // bug doesn't ship a capsule the peer must reject.
+    require(reasonBytes.size <= MAX_WT_CLOSE_REASON_BYTES) {
+        "WT_CLOSE_SESSION reason exceeds $MAX_WT_CLOSE_REASON_BYTES bytes (${reasonBytes.size})"
+    }
     val body = QuicWriter()
     body.writeUint32(errorCode)
-    body.writeBytes(reason.encodeToByteArray())
+    body.writeBytes(reasonBytes)
     return encodeCapsule(WtCapsuleType.WT_CLOSE_SESSION, body.toByteArray())
 }
 
+/** draft-ietf-webtrans-http3 §5 cap on the reason string in WT_CLOSE_SESSION. */
+const val MAX_WT_CLOSE_REASON_BYTES: Int = 8192
+
 /** Parsed WT_CLOSE_SESSION capsule. */
 data class WtCloseSession(
     val errorCode: Int,
@@ -64,11 +74,22 @@ data class WtCloseSession(
  * Stateful capsule reader. Capsules on the WT CONNECT bidi stream are
  * `(varint type)(varint length)(body)`. Feed bytes via [push]; drain
  * complete capsules via [next].
+ *
+ * Memory safety: pending unparsed bytes are capped at [maxPendingBytes]
+ * and per-capsule body size is capped at [maxCapsuleBodyBytes]. A peer
+ * that streams a partial capsule prefix (or advertises a 4 GiB length)
+ * cannot pin unbounded heap.
  */
-class CapsuleReader {
+class CapsuleReader(
+    private val maxPendingBytes: Int = DEFAULT_MAX_PENDING_BYTES,
+    private val maxCapsuleBodyBytes: Int = DEFAULT_MAX_CAPSULE_BODY_BYTES,
+) {
     private var buf: ByteArray = ByteArray(0)
     private var pos: Int = 0
 
+    /** Bytes currently buffered awaiting a complete capsule. Diagnostic. */
+    val bufferedBytes: Int get() = buf.size - pos
+
     fun push(bytes: ByteArray) {
         if (bytes.isEmpty()) return
         // Amortized compaction (same pattern as Http3FrameReader).
@@ -76,6 +97,14 @@ class CapsuleReader {
             buf = buf.copyOfRange(pos, buf.size)
             pos = 0
         }
+        val newPending = (buf.size - pos).toLong() + bytes.size.toLong()
+        if (newPending > maxPendingBytes) {
+            throw com.vitorpamplona.quic.QuicCodecException(
+                "WT capsule reader buffer would exceed cap " +
+                    "($newPending > $maxPendingBytes); peer is streaming a partial " +
+                    "capsule without delivering the body",
+            )
+        }
         val combined = ByteArray(buf.size + bytes.size)
         buf.copyInto(combined, 0)
         bytes.copyInto(combined, buf.size)
@@ -97,8 +126,10 @@ class CapsuleReader {
                 .decode(buf, typeEnd) ?: return null
         val bodyStart = typeEnd + lenRes.bytesConsumed
         val len = lenRes.value
-        if (len < 0 || len > Int.MAX_VALUE.toLong()) {
-            throw com.vitorpamplona.quic.QuicCodecException("capsule length out of range: $len")
+        if (len < 0 || len > maxCapsuleBodyBytes.toLong()) {
+            throw com.vitorpamplona.quic.QuicCodecException(
+                "capsule length out of range: $len (cap $maxCapsuleBodyBytes)",
+            )
         }
         val bodyEnd = bodyStart + len.toInt()
         if (bodyEnd > buf.size) return null
@@ -137,4 +168,16 @@ class CapsuleReader {
             }
         }
     }
+
+    companion object {
+        /** Per-stream pending-capsule cap (1 MiB). */
+        const val DEFAULT_MAX_PENDING_BYTES: Int = 1 shl 20
+
+        /**
+         * Per-capsule body cap (64 KiB). The largest legitimate capsule
+         * we expect is WT_CLOSE_SESSION (≤ [MAX_WT_CLOSE_REASON_BYTES] +
+         * 4 bytes); 64 KiB is generous headroom for future capsule types.
+         */
+        const val DEFAULT_MAX_CAPSULE_BODY_BYTES: Int = 64 * 1024
+    }
 }

quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtDatagram.kt

@@ -45,15 +45,28 @@ object WtDatagram {
         return w.toByteArray()
     }
 
-    /** Returns (quarterStreamId * 4, payload). Returns null on truncation. */
+    /** Returns (quarterStreamId * 4, payload). Returns null on truncation or out-of-range quarter id. */
     fun decode(bytes: ByteArray): Decoded? {
         val r = Varint.decode(bytes, 0) ?: return null
         if (r.bytesConsumed > bytes.size) return null
+        // QUIC stream ids are bounded by 2^62 - 1 (RFC 9000 §16). The
+        // quarter id is the actual stream id divided by 4, so a valid
+        // wire value is at most (2^62 - 1) / 4. Anything larger is
+        // either malformed or a deliberate attempt to overflow `r.value
+        // * 4` past `Long.MAX_VALUE` and have it wrap into a small
+        // signed value that matches our expected session id — which
+        // would let an off-path peer feed forged datagrams to the
+        // application bypassing the session-id check in
+        // [QuicWebTransportSessionState.pollIncomingDatagram].
+        if (r.value < 0 || r.value > MAX_VALID_QUARTER_ID) return null
         val sessionId = r.value * 4
         val payload = bytes.copyOfRange(r.bytesConsumed, bytes.size)
         return Decoded(sessionId, payload)
     }
 
+    /** (2^62 - 1) / 4 — see [decode]. */
+    internal const val MAX_VALID_QUARTER_ID: Long = ((1L shl 62) - 1L) / 4L
+
     data class Decoded(
         val sessionStreamId: Long,
         val payload: ByteArray,

quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt

@@ -82,8 +82,20 @@ class WtPeerStreamDemux(
      * can leave this null.
      */
     private val driver: com.vitorpamplona.quic.connection.QuicConnectionDriver? = null,
+    /**
+     * Cap on the number of peer-initiated streams that may queue here
+     * waiting for the application to consume them. Pre-fix this was
+     * `Channel.UNLIMITED` — a peer that opens streams faster than the
+     * application drains could pin one [StrippedWtStream] (and its
+     * captured `chunkChannel`) per stream id, indefinitely. Bounded
+     * buffer + suspending `send` propagates backpressure: when the app
+     * is slow, [route] suspends inside [emitStripped] before launching
+     * any further per-stream work, which in turn lets QUIC's per-conn
+     * `MAX_STREAMS_*` accounting throttle the peer.
+     */
+    private val readyStreamsBuffer: Int = DEFAULT_READY_STREAMS_BUFFER,
 ) {
-    private val readyStreams = Channel<StrippedWtStream>(Channel.UNLIMITED)
+    private val readyStreams = Channel<StrippedWtStream>(readyStreamsBuffer)
 
     @Volatile
     var peerSettings: Http3Settings? = null
@@ -113,6 +125,21 @@ class WtPeerStreamDemux(
     var peerGoawayProtocolError: String? = null
         private set
 
+    /**
+     * Surface RFC 9114 §7.2 frame-validation violations the
+     * [Http3FrameReader] catches (H3_FRAME_UNEXPECTED, H3_MISSING_SETTINGS,
+     * forbidden reserved types). The route()-level `catch (_: Throwable)`
+     * cancels the per-stream collector but doesn't propagate the
+     * underlying [com.vitorpamplona.quic.QuicCodecException]'s message
+     * upstream — without this field a buggy server would just see
+     * "stream stops being read" with no diagnostic. Stays null until a
+     * violation is observed; the QUIC-layer caller polls it and closes
+     * the connection on non-null.
+     */
+    @Volatile
+    var peerH3ProtocolError: String? = null
+        private set
+
     val incomingStrippedStreams: Flow<StrippedWtStream> = readyStreams.consumeAsFlow()
 
     /**
@@ -134,7 +161,16 @@ class WtPeerStreamDemux(
         kotlinx.coroutines.coroutineScope {
             val pending = ArrayDeque<ByteArray>()
             val flowIterator = stream.incoming
-            val chunkChannel = Channel<ByteArray>(Channel.UNLIMITED)
+            // Bounded suspending channel — pre-fix `UNLIMITED` let a
+            // single misbehaving peer stream pin gigabytes of heap when
+            // the consumer (the application) couldn't keep up. With a
+            // bounded buffer + suspending `send`, the collector below
+            // back-pressures the QuicStream's incoming flow — which in
+            // turn delays our outbound MAX_STREAM_DATA grants and
+            // throttles the peer at the QUIC layer. The fixed size is
+            // small (64 chunks ≈ ~64 KiB at typical packet payloads):
+            // anything bigger just delays the back-pressure signal.
+            val chunkChannel = Channel<ByteArray>(CHUNK_CHANNEL_BUFFER)
             val collector =
                 launch {
                     try {
@@ -225,13 +261,25 @@ class WtPeerStreamDemux(
         pending: ArrayDeque<ByteArray>,
         chunkChannel: Channel<ByteArray>,
     ) {
-        val reader = Http3FrameReader()
+        val reader = Http3FrameReader(context = Http3FrameReader.StreamContext.CONTROL)
         // Push whatever we already buffered.
-        while (pending.isNotEmpty()) reader.push(pending.removeFirst())
-        consumeFrames(reader)
-        for (chunk in chunkChannel) {
-            reader.push(chunk)
+        try {
+            while (pending.isNotEmpty()) reader.push(pending.removeFirst())
             consumeFrames(reader)
+            for (chunk in chunkChannel) {
+                reader.push(chunk)
+                consumeFrames(reader)
+            }
+        } catch (e: com.vitorpamplona.quic.QuicCodecException) {
+            // RFC 9114 §7.2 frame-validation throws (H3_FRAME_UNEXPECTED /
+            // H3_MISSING_SETTINGS / reserved type) land here. Record the
+            // diagnostic so the QUIC layer / application can close the
+            // connection deliberately instead of having the route() catch
+            // silently swallow the message. Idempotent on duplicate hits.
+            if (peerH3ProtocolError == null) {
+                peerH3ProtocolError = e.message ?: "HTTP/3 protocol violation on CONTROL stream"
+            }
+            throw e
         }
     }
 
@@ -289,7 +337,7 @@ class WtPeerStreamDemux(
         }
     }
 
-    private fun emitStripped(
+    private suspend fun emitStripped(
         stream: QuicStream,
         pending: ArrayDeque<ByteArray>,
         chunkChannel: Channel<ByteArray>,
@@ -324,7 +372,14 @@ class WtPeerStreamDemux(
                     driver?.wakeup()
                 }
             }
-        readyStreams.trySend(
+        // Suspending send instead of `trySend` so a slow application
+        // back-pressures the demux instead of silently dropping the
+        // stream. With a bounded buffer ([readyStreamsBuffer]),
+        // [readyStreams.send] suspends when the app hasn't drained,
+        // which keeps `route()` busy and (combined with [process]'s
+        // `scope.launch`) lets the next peer stream wait its turn
+        // rather than spawning a coroutine for every offered id.
+        readyStreams.send(
             StrippedWtStream(
                 streamId = stream.streamId,
                 isUnidirectional = isUni,
@@ -334,6 +389,14 @@ class WtPeerStreamDemux(
             ),
         )
     }
+
+    companion object {
+        /** Default cap on queued peer-initiated streams (1024). */
+        const val DEFAULT_READY_STREAMS_BUFFER: Int = 1024
+
+        /** Per-stream chunk-channel capacity (64). */
+        internal const val CHUNK_CHANNEL_BUFFER: Int = 64
+    }
 }
 
 private fun flatten(chunks: ArrayDeque<ByteArray>): ByteArray {

quic/src/commonTest/kotlin/com/vitorpamplona/quic/http3/Http3FrameReaderTest.kt

@@ -20,10 +20,13 @@
  */
 package com.vitorpamplona.quic.http3
 
+import com.vitorpamplona.quic.QuicCodecException
+import com.vitorpamplona.quic.QuicWriter
 import com.vitorpamplona.quic.webtransport.encodeHeadersFrame
 import kotlin.test.Test
 import kotlin.test.assertContentEquals
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 import kotlin.test.assertNull
 import kotlin.test.assertTrue
 
@@ -90,4 +93,55 @@ class Http3FrameReaderTest {
         assertContentEquals(byteArrayOf(0x01, 0x02, 0x03, 0x04), second.body)
         assertNull(r.next())
     }
+
+    @Test
+    fun control_stream_first_frame_must_be_settings() {
+        // RFC 9114 §7.2.4: first CONTROL-stream frame is SETTINGS or
+        // H3_MISSING_SETTINGS. A peer that opens a CONTROL stream and
+        // immediately sends GOAWAY is malformed.
+        val r = Http3FrameReader(context = Http3FrameReader.StreamContext.CONTROL)
+        val w = QuicWriter()
+        w.writeVarint(Http3FrameType.GOAWAY)
+        w.writeVarint(0L)
+        r.push(w.toByteArray())
+        assertFailsWith<QuicCodecException> { r.next() }
+    }
+
+    @Test
+    fun control_stream_rejects_data_frame() {
+        // RFC 9114 §7.2.1: DATA frames are forbidden on the control stream.
+        val r = Http3FrameReader(context = Http3FrameReader.StreamContext.CONTROL)
+        // Settings first (legal), then DATA (illegal).
+        r.push(buildClientWebTransportSettings().encodeFrame())
+        assertTrue(r.next() is Http3Frame.Settings)
+        val w = QuicWriter()
+        w.writeVarint(Http3FrameType.DATA)
+        w.writeVarint(0L)
+        r.push(w.toByteArray())
+        assertFailsWith<QuicCodecException> { r.next() }
+    }
+
+    @Test
+    fun request_stream_rejects_settings_frame() {
+        // RFC 9114 §7.2.4: SETTINGS forbidden on request streams.
+        val r = Http3FrameReader(context = Http3FrameReader.StreamContext.REQUEST)
+        r.push(buildClientWebTransportSettings().encodeFrame())
+        assertFailsWith<QuicCodecException> { r.next() }
+    }
+
+    @Test
+    fun reserved_frame_types_are_rejected() {
+        // RFC 9114 §7.2.8: types 0x02, 0x06, 0x08, 0x09 are reserved
+        // and MUST be treated as connection error.
+        for (forbidden in longArrayOf(0x02L, 0x06L, 0x08L, 0x09L)) {
+            val r = Http3FrameReader(context = Http3FrameReader.StreamContext.REQUEST)
+            val w = QuicWriter()
+            w.writeVarint(forbidden)
+            w.writeVarint(0L)
+            r.push(w.toByteArray())
+            assertFailsWith<QuicCodecException>("type 0x${forbidden.toString(16)} must be rejected") {
+                r.next()
+            }
+        }
+    }
 }

quic/src/commonTest/kotlin/com/vitorpamplona/quic/qpack/HuffmanRfc7541Test.kt

@@ -94,4 +94,27 @@ class HuffmanRfc7541Test {
         val decoded = QpackHuffman.decode(encoded).decodeToString()
         assertEquals("https://www.example.com", decoded)
     }
+
+    @Test
+    fun length_30_codes_decode_correctly() {
+        // RFC 7541 Appendix B: symbols 10 (LF), 13 (CR), 22 (DC2) all use
+        // 30-bit codes. Hand-encoded with two trailing pad-1 bits so the
+        // total spans exactly 4 bytes. Pre-fix the [codesByLen] range
+        // accidentally excluded length 30 and these three bytes were
+        // silently rejected as "invalid Huffman bit stream".
+        // sym 10 (LF):  code 0x3FFFFFFC | (pad 11) → FF FF FF F3
+        // sym 13 (CR):  code 0x3FFFFFFD | (pad 11) → FF FF FF F7
+        // sym 22 (DC2): code 0x3FFFFFFE | (pad 11) → FF FF FF FB
+        val cases =
+            listOf(
+                10 to byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0xFF.toByte(), 0xF3.toByte()),
+                13 to byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0xFF.toByte(), 0xF7.toByte()),
+                22 to byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0xFF.toByte(), 0xFB.toByte()),
+            )
+        for ((sym, encoded) in cases) {
+            val decoded = QpackHuffman.decode(encoded)
+            assertEquals(1, decoded.size, "sym=$sym")
+            assertEquals(sym.toByte(), decoded[0], "sym=$sym")
+        }
+    }
 }

quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/crypto/JcaAesGcmAead.kt

@@ -34,8 +34,12 @@ import javax.crypto.spec.SecretKeySpec
  * (which is much cheaper than `getInstance`) plus the AEAD math itself.
  *
  * Single-thread per direction: one PacketProtection feeds either the read
- * loop OR the send loop, never both. Locking would be needed if that ever
- * changes.
+ * loop OR the send loop, never both. The class still synchronizes on a
+ * private monitor as a defence-in-depth: the lock-split refactor in
+ * `QuicConnectionDriver` keeps each side single-threaded by design, but
+ * a future caller (test harness, key-update path) sharing the instance
+ * across coroutines would otherwise corrupt the cached `Cipher` state
+ * silently. The JCA `Cipher` itself is documented as not thread-safe.
  */
 class JcaAesGcmAead(
     key: ByteArray,
@@ -58,29 +62,63 @@ class JcaAesGcmAead(
     // Cipher.getInstance — slow but rare (once per Initial datagram).
     private val encryptCipher: Cipher = Cipher.getInstance("AES/GCM/NoPadding")
     private val decryptCipher: Cipher = Cipher.getInstance("AES/GCM/NoPadding")
-    private var lastEncryptNonce: ByteArray? = null
+
+    /**
+     * Last nonce successfully consumed by [seal]. We use a fresh
+     * [Cipher.getInstance] when the caller asks us to seal under the
+     * SAME nonce a second time (RFC 9000 §14 Initial-padding rebuild).
+     *
+     * Recent-history set rather than just the most-recent nonce: a
+     * single intermediate seal between two rebuilds could otherwise
+     * mask a duplicate against the SECOND-most-recent nonce, which
+     * some JCA providers (Conscrypt) reject with
+     * `InvalidAlgorithmParameterException` while others (SunJCE)
+     * silently allow. Bounded at [NONCE_HISTORY_LIMIT] entries — far
+     * more than any legitimate rebuild path needs (typically 1–2),
+     * but cheap to keep.
+     */
+    private val recentEncryptNonces = ArrayDeque<ByteArray>()
 
     override fun seal(
         key: ByteArray,
         nonce: ByteArray,
         aad: ByteArray,
         plaintext: ByteArray,
-    ): ByteArray {
-        // Detect IV reuse — happens on the Initial-padding rebuild path. Fall
-        // back to a one-shot fresh Cipher for that case rather than fighting
-        // JCA's safety check.
-        val reuse = lastEncryptNonce?.contentEquals(nonce) == true
-        return if (reuse) {
-            val fresh = Cipher.getInstance("AES/GCM/NoPadding")
-            fresh.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
-            fresh.updateAAD(aad)
-            fresh.doFinal(plaintext)
-        } else {
-            encryptCipher.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
-            encryptCipher.updateAAD(aad)
-            val out = encryptCipher.doFinal(plaintext)
-            lastEncryptNonce = nonce
-            out
+    ): ByteArray =
+        synchronized(this) {
+            val reuse = recentEncryptNonces.any { it.contentEquals(nonce) }
+            if (reuse) {
+                val fresh = Cipher.getInstance("AES/GCM/NoPadding")
+                fresh.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
+                fresh.updateAAD(aad)
+                fresh.doFinal(plaintext)
+            } else {
+                try {
+                    encryptCipher.init(Cipher.ENCRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
+                    encryptCipher.updateAAD(aad)
+                    val out = encryptCipher.doFinal(plaintext)
+                    rememberEncryptNonce(nonce)
+                    out
+                } catch (t: Throwable) {
+                    // Any throw mid-`init`/`updateAAD`/`doFinal` leaves the
+                    // cached cipher in a provider-defined state. The next
+                    // legitimate call's `init` should reset it, but if a
+                    // crafted input triggers a partial init the next seal
+                    // could conceivably reuse residual state. Drop the
+                    // most recent nonce from the history so a retry with
+                    // the same nonce takes the safe fresh-Cipher path.
+                    if (recentEncryptNonces.lastOrNull()?.contentEquals(nonce) == true) {
+                        recentEncryptNonces.removeLast()
+                    }
+                    throw t
+                }
+            }
+        }
+
+    private fun rememberEncryptNonce(nonce: ByteArray) {
+        recentEncryptNonces.addLast(nonce)
+        while (recentEncryptNonces.size > NONCE_HISTORY_LIMIT) {
+            recentEncryptNonces.removeFirst()
         }
     }
 
@@ -90,11 +128,17 @@ class JcaAesGcmAead(
         aad: ByteArray,
         ciphertext: ByteArray,
     ): ByteArray? =
-        try {
-            decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
-            decryptCipher.updateAAD(aad)
-            decryptCipher.doFinal(ciphertext)
-        } catch (_: GeneralSecurityException) {
-            null
+        synchronized(this) {
+            try {
+                decryptCipher.init(Cipher.DECRYPT_MODE, keySpec, GCMParameterSpec(128, nonce))
+                decryptCipher.updateAAD(aad)
+                decryptCipher.doFinal(ciphertext)
+            } catch (_: GeneralSecurityException) {
+                null
+            }
         }
+
+    private companion object {
+        private const val NONCE_HISTORY_LIMIT = 8
+    }
 }

quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt

@@ -165,10 +165,23 @@ actual class UdpSocket private constructor(
                     channel.setOption(StandardSocketOptions.IP_TOS, ECT0_TOS_BITS)
                 }
                 channel.bind(InetSocketAddress(0)) // ephemeral
-                // We use receive()/send(addr) instead of channel.connect() so that
-                // sendDatagram-style flows can still be implemented on the same
-                // socket if we ever need them. For the pure client use-case this
-                // is identical in latency.
+                // RFC 9000 §9 / defence-in-depth: connect() asks the kernel
+                // to filter inbound datagrams to those from [remote]. Without
+                // this any host on the network can spoof our 4-tuple and
+                // force us to attempt AEAD decryption on garbage — each
+                // failed packet costs a `Cipher.init` + AAD/decrypt + tag
+                // check, and a successful unauthenticated stateless reset
+                // forgery would terminate our connection. With connect(),
+                // the kernel rejects mismatched-source datagrams before
+                // they reach userspace.
+                //
+                // We connect AFTER bind so the ephemeral local port is
+                // chosen first, then the destination is associated. Pure
+                // client semantics — `quic` doesn't currently support
+                // server-side or connection migration to a new remote IP
+                // (path validation rotates DCIDs on the SAME 4-tuple), so
+                // pinning the socket to one remote is a strict win.
+                channel.connect(remote)
                 UdpSocket(channel, remote)
             }
     }