fix(quic): stabilization pass — concurrency, RFC compliance, DoS hardening

Verified and applied 12 focused fixes from a four-agent review of the quic module. Each fix verified against the actual code; agent findings that traced to false positives (pendingPing clear-without-emit, sentPackets on encrypt failure) are documented in the review thread but not changed. Concurrency / flakiness: - PTO consecutive count: double-increment removed; now incremented exactly once per PTO event in handlePtoFired before requeue, so the threshold check inside requeueInflightForProbe sees the post- increment value AND the between-probe re-requeue doesn't bump again. - Wallclock → monotonic: QuicConnection.nowMillis defaults to TimeSource.Monotonic anchored at construction, so NTP step / suspend-resume can't poison RTT samples. Driver uses connection.nowMillis instead of carrying its own clock. - Close-state race: atomic CAS via closeStateMonitor in close() and markClosedExternally so concurrent teardown paths can't both fire qlog "connection closed" and stomp on closeReason. - streamsList CME: converted to @Volatile var List<QuicStream> with immutable-snapshot publishing under streamsLock. closeAllSignals' iteration is now CME-free without holding the (suspending) lock. - JcaAesGcmAead: synchronized seal/open; multi-entry recent-nonce history (was single most-recent — could mask a duplicate against the second-most-recent under intermediate seals); on seal failure evict the cached nonce so a retry with the same nonce takes the safe fresh-Cipher path. Wire correctness / spec: - Connection-level send credit no longer debited on retransmits (added Chunk.isRetransmit; writer skips sendConnectionFlowConsumed += data.size when set). Pre-fix a few PTO rounds on a long stream exhausted credit and stalled the connection. - ACK-delay shl overflow: clamp ackDelayExponent to 0..20 and clamp the peer's varint to (Long.MAX_VALUE >>> exponent) before shift; clamp negative now-vs-recv-time before shift on outbound AckTracker. - Key-update commit only when new-phase packet PN exceeds largestReceived (RFC 9001 §6.1). - RESET_STREAM final-size validation: enforce equality with prior FIN size and ≥ highestObservedOffset; close FINAL_SIZE_ERROR otherwise. - STOP_SENDING handling: respond with RESET_STREAM on the local send side (was silently dropped, peer's flow-credit wasted). - ReceiveBuffer.insert: typed InsertResult; second FIN with conflicting size and offset-past-FIN data both surface FINAL_SIZE_ERROR via the parser instead of being silently dropped. - Retry SCID==DCID self-loop: reject Retry where the peer's SCID equals our original DCID (RFC 9000 §17.2.5.2). DoS hardening: - ACK PN walk: drainAckedSentPackets rewritten to scan in-flight keys against parsed ranges instead of walking every PN. A peer with firstAckRange = 2^62-1 used to pin a core forever; now bounded by the sent-packets map size. - UDP socket: channel.connect(remote) after bind so the kernel filters off-path datagrams. Stops trivial source spoofing from burning AEAD attempts. https://claude.ai/code/session_01AhGvbMV8uPRse3TmAGaddM
2026-05-08 22:56:05 +00:00
parent fbadc61070
commit b25644bf8b
10 changed files with 486 additions and 120 deletions
@@ -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,
@@ -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
@@ -408,7 +408,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 +508,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 +623,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 +745,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 -> {
@@ -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
@@ -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
+}
@@ -143,8 +143,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,
@@ -54,27 +54,43 @@ 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
+        if (end <= readOffset) return InsertResult.OK
        val effOffset: Long
        val effData: ByteArray
        if (offset < readOffset) {
@@ -101,7 +117,7 @@ class ReceiveBuffer {
        if (startIdx == endIdx) {
            // No overlap — just insert.
            chunks.add(startIdx, Chunk(effOffset, effData))
-            return
+            return InsertResult.OK
        }

        // Coalesce [startIdx, endIdx) plus the new chunk.
@@ -119,6 +135,32 @@ class ReceiveBuffer {
        // Replace
        for (i in 1..(endIdx - startIdx)) chunks.removeAt(startIdx)
        chunks.add(startIdx, Chunk(lo, merged))
+        return InsertResult.OK
+    }
+
+    /** 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]. */
@@ -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
        }
@@ -625,21 +625,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
        }
@@ -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
+    }
 }
@@ -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)
            }
    }