fix(quic): round-5 audit fixes — concurrency, ack-eliciting flags, scope leaks

Two parallel audit agents inspected the round-4 commits for regressions and concurrency hazards. Major findings: ackEliciting regression (HIGH from core-regression report): Round-4's ACK gating optimization (only emit ACKs when something ack-eliciting was received) didn't update the parser's per-frame handling. MaxDataFrame, MaxStreamDataFrame, MaxStreamsFrame, NewConnectionIdFrame, HandshakeDoneFrame, ResetStreamFrame, StopSendingFrame, NewTokenFrame all need ackEliciting=true per RFC 9000 §13.2.1. Pre-fix a packet carrying only one of these would record the PN but never trigger an ACK, causing the peer to PTO-retransmit forever. HandshakeDoneFrame conditional (HIGH): Pre-fix the dispatcher unconditionally set status=CONNECTED; if applyPeerTransportParameters had just called markClosedExternally (e.g. CID-validation failure), a later HANDSHAKE_DONE in the same payload would resurrect the connection. Now only sets CONNECTED when status is HANDSHAKING. WT scope leak (CRITICAL from concurrency report): QuicWebTransportSessionState.close() never cancelled the scope holding the demux pump and capsule reader coroutines; both kept running past close, retaining QuicStream / chunk channels indefinitely. Memory growth on long sessions that opened/closed many WT sessions. WtPeerStreamDemux.route() collector leak (CRITICAL): The route function launches a coroutine to drain stream.incoming into chunkChannel (UNLIMITED). Four early-return paths (truncated stream type, mismatched WT signal, foreign session id) returned without closing chunkChannel — collector kept running, channel grew unbounded. Now wrapped in coroutineScope{} so the collector is joined on every exit. Also explicitly cancels collector on the catch path. RESET_STREAM stream-id ownership (HIGH): Pre-fix the dispatcher closed the local read side on whatever stream the peer named. RFC 9000 §3.5: the peer can only RESET_STREAM streams where it owns a send side. A peer RESETting a CLIENT_UNI is STREAM_STATE_ERROR (we own the only side). Now closes the connection in that case. looksLikeIpLiteral tightening (HIGH): Pre-fix accepted "1.2.3.4.5", "1.2", "1." as IP literals — Java's InetAddress.getByName resolves all of those via DNS, defeating audit-4 #4's SNI-leak fix. Now strict: 4 dot-separated octets each in 0..255, or contains a colon (IPv6). Signal channels closed on teardown (MEDIUM): closeAllSignals() helper closes peerStreamSignal + incomingDatagramSignal alongside closedSignal; pre-fix only closedSignal was closed and racing parser frames could still trySend into never-consumed channels. Centralised the call so close() and markClosedExternally both invoke it. Driver.close() idempotency (HIGH): A second concurrent close() (common: session close + read-loop death racing) used to launch a parallel teardown that called scope.cancel() while the first's joinAll was mid-flight. Now memoizes the launched Job behind a synchronized block. Driver.close() flush detection (MEDIUM): Pre-fix spun on `pendingDatagrams.isEmpty()` to detect CONNECTION_CLOSE flush, but the writer's CLOSING branch bypasses pendingDatagrams entirely. Now spins on `connection.status == CLOSING`, which transitions to CLOSED only after drainOutbound builds the close packet. @Volatile on peerMaxStreams* (MEDIUM): peerMaxStreamsBidi/Uni snapshots are documented lock-free; without @Volatile, JLS allows long-tearing on 32-bit JVMs and the JIT may cache stale values. CertificateFactory parse inside try (MEDIUM): Malformed cert chain bytes used to throw raw CertificateException through the read loop. Now wrapped, so parse failure becomes a clean CONNECTION_CLOSE. GOAWAY id-regression observability (MEDIUM): Pre-fix the QuicCodecException thrown on increasing GOAWAY id was silently swallowed by route()'s catch. Now also surfaces via peerGoawayProtocolError so the application/QUIC layer can act. appendFlowControlUpdates uses streamsListLocked (perf): The round-4 perf #10 fix introduced streamsListLocked (no entries.toList per drain) but appendFlowControlUpdates still iterated the Map. Now also uses the index-friendly view. peerCloseDeferred completion on session close (MEDIUM): awaitPeerClose() used to hang forever if the local side called close() before any peer-initiated WT_CLOSE_SESSION arrived. Now cancelled with CancellationException on local close. Tests: AckElicitingFramesTest pins the ackEliciting contract on every round-5 fix plus the ResetStream-on-CLIENT_UNI rejection. JdkCertificateValidatorIpLiteralTest pins the tightened pattern via reflection. https://claude.ai/code/session_01EC1tfXfap8k8GyKvrxkxZx
2026-04-26 01:06:52 +00:00
parent 920b36cdd6
commit 7f05fd6e2a
9 changed files with 560 additions and 140 deletions
@@ -122,8 +122,16 @@ class QuicConnection(
     *
     * [openBidiStream] consults this cap; opening past it would violate the
     * peer's flow control and trigger STREAM_LIMIT_ERROR on their side.
+     *
+     * Round-5 concurrency #7: `@Volatile` because [peerMaxStreamsBidiSnapshot]
+     * is documented as lock-free; without volatile, JLS allows long-tearing on
+     * 32-bit JVMs (still common on Android) and the JIT may cache a stale
+     * value indefinitely.
     */
+    @Volatile
    internal var peerMaxStreamsBidi: Long = 0L
+
+    @Volatile
    internal var peerMaxStreamsUni: Long = 0L

    /**
@@ -453,7 +461,7 @@ class QuicConnection(
        if (!handshakeComplete) {
            signalHandshakeFailed(QuicConnectionClosedException("connection closed before handshake completed: $reason"))
        }
-        closedSignal.close()
+        closeAllSignals()
    }

    /** Called by the parser on inbound CONNECTION_CLOSE or by the driver on read-loop death. */
@@ -462,7 +470,21 @@ class QuicConnection(
        if (!handshakeComplete) {
            signalHandshakeFailed(QuicConnectionClosedException("connection closed externally: $reason"))
        }
+        closeAllSignals()
+    }
+
+    /**
+     * Close every wakeup channel so suspended awaiters exit promptly. Round-5
+     * concurrency #11: closing only `closedSignal` left `peerStreamSignal` and
+     * `incomingDatagramSignal` open, so a parser frame racing teardown could
+     * still `trySend(Unit)` into a never-consumed channel. All three channels
+     * close idempotently, so calling this from both `close()` and
+     * `markClosedExternally` is safe.
+     */
+    private fun closeAllSignals() {
        closedSignal.close()
+        peerStreamSignal.close()
+        incomingDatagramSignal.close()
    }

    /**
@@ -61,6 +61,16 @@ class QuicConnectionDriver(
    private var readJob: Job? = null
    private var sendJob: Job? = null

+    /**
+     * Round-5 concurrency #5: close() guard. A second concurrent invocation
+     * (e.g. session close + read-loop death close racing) used to launch a
+     * parallel teardown that called scope.cancel() and socket.close() while
+     * the first close was mid-joinAll. We now memoize the teardown Job so
+     * the second caller awaits the first's completion instead.
+     */
+    @Volatile
+    private var closeJob: Job? = null
+
    fun start() {
        connection.start()
        readJob = scope.launch { readLoop() }
@@ -147,45 +157,42 @@ class QuicConnectionDriver(
     * datagrams or skipping the CONNECTION_CLOSE entirely.
     */
    fun close() {
-        parentScope.launch {
-            connection.close(0L, "")
-            wakeup()
-            val send = sendJob
-            // Bounded wait for the send loop to flush CONNECTION_CLOSE. We
-            // don't want to hang forever if the writer is wedged — the timeout
-            // is the upper bound on how long close() can block.
-            withTimeoutOrNull(CLOSE_FLUSH_TIMEOUT_MILLIS) {
-                // Spin until the writer has actually drained the queued close
-                // (queues are empty AND the send loop has cycled at least
-                // once). Easiest proxy: write was attempted and there's
-                // nothing more to send. We approximate by giving the loop a
-                // chance to drain by sleeping briefly. This is the one place
-                // a short sleep is acceptable because we're racing a flush.
-                while (true) {
-                    val drained =
-                        connection.lock.withLock {
-                            // No more pending datagrams or stream bytes? Then
-                            // CONNECTION_CLOSE has either been sent or there
-                            // was nothing to send.
-                            connection.pendingDatagramsLocked().isEmpty()
+        // Round-5 #5: idempotent close. Memoize the teardown launch so a
+        // second concurrent caller (which is common: session.close() and
+        // read-loop death both race to close()) awaits the same Job rather
+        // than launching a parallel teardown.
+        if (closeJob != null) return
+        synchronized(this) {
+            if (closeJob != null) return
+            closeJob =
+                parentScope.launch {
+                    connection.close(0L, "")
+                    wakeup()
+                    val send = sendJob
+                    // Bounded wait for the send loop to flush CONNECTION_CLOSE.
+                    // We don't want to hang forever if the writer is wedged —
+                    // the timeout is the upper bound on how long close() blocks.
+                    withTimeoutOrNull(CLOSE_FLUSH_TIMEOUT_MILLIS) {
+                        // Spin until the writer has actually drained the queued
+                        // close. The CLOSING-status check transitions to CLOSED
+                        // once drainOutbound builds the CONNECTION_CLOSE packet.
+                        while (connection.status == QuicConnection.Status.CLOSING) {
+                            kotlinx.coroutines.delay(1)
                        }
-                    if (drained) break
-                    kotlinx.coroutines.delay(1)
+                    }
+                    // Now flip to CLOSED so both loops exit their while-guards.
+                    connection.markClosedExternally("driver close requested")
+                    wakeup()
+                    // Wait for both loops to actually exit — joinAll won't
+                    // return until the in-flight socket.send() completes.
+                    withTimeoutOrNull(CLOSE_FLUSH_TIMEOUT_MILLIS) {
+                        listOfNotNull(readJob, send).joinAll()
+                    }
+                    // Final teardown — cancel guarantees both jobs are done
+                    // before we close the socket.
+                    scope.cancel()
+                    socket.close()
                }
-            }
-            // Now flip to CLOSED so both loops exit their while-guards.
-            connection.markClosedExternally("driver close requested")
-            wakeup()
-            // Wait for both loops to actually exit — joinAll won't return
-            // until the in-flight socket.send() (if any) completes.
-            withTimeoutOrNull(CLOSE_FLUSH_TIMEOUT_MILLIS) {
-                listOfNotNull(readJob, send).joinAll()
-            }
-            // Final teardown. By now both jobs have either exited cleanly or
-            // exceeded the timeout — cancel guarantees they're done before we
-            // close the socket.
-            scope.cancel()
-            socket.close()
        }
    }

@@ -264,8 +264,10 @@ private fun dispatchFrames(
            }

            is MaxDataFrame -> {
-                // Audit-4 #9 + #12: previously a no-op, which silently stalled
-                // the writer once we sent more than initialMaxData total bytes.
+                // RFC 9000 §13.2.1: MAX_DATA is ack-eliciting. Without this,
+                // a packet carrying only MAX_DATA would record the PN but
+                // never trigger an ACK (round-4 ACK gating regression).
+                ackEliciting = true
                // RFC 9000 §19.9: MAX_DATA only ever raises the cap.
                if (frame.maxData > conn.sendConnectionFlowCredit) {
                    conn.sendConnectionFlowCredit = frame.maxData
@@ -273,12 +275,14 @@ private fun dispatchFrames(
            }

            is MaxStreamDataFrame -> {
+                ackEliciting = true
                conn.streamByIdLocked(frame.streamId)?.let {
                    if (frame.maxStreamData > it.sendCredit) it.sendCredit = frame.maxStreamData
                }
            }

            is MaxStreamsFrame -> {
+                ackEliciting = true
                // RFC 9000 §19.11: MAX_STREAMS only ever raises the cap.
                // Frames with values smaller than the current cap are ignored.
                // Bidi vs uni is signaled via the frame's `bidi` flag.
@@ -294,31 +298,42 @@ private fun dispatchFrames(
            }

            is ResetStreamFrame -> {
-                // Audit-4 #2: peer aborted the send side of a stream. We
-                // accept the frame for survival; the receive buffer keeps
-                // whatever it had. Closing the local read flow is left to
-                // the application or a future enhancement that surfaces the
-                // application error code.
+                // RFC 9000 §3.5: RESET_STREAM is the peer aborting THEIR send
+                // side of the stream. Round-5 #2: it's only legal on streams
+                // where the peer owns a send side (server-initiated streams,
+                // or our own bidi). A peer RESETting one of OUR uni streams
+                // (CLIENT_UNI = id%4==2) is STREAM_STATE_ERROR — they don't
+                // have a send side to abort.
                ackEliciting = true
+                if (StreamId.kindOf(frame.streamId) == StreamId.Kind.CLIENT_UNI) {
+                    conn.markClosedExternally(
+                        "STREAM_STATE_ERROR: peer RESET_STREAM on client-uni id ${frame.streamId} (peer has no send side)",
+                    )
+                    return
+                }
+                // Mark the peer's stream aborted and close our read side; the
+                // application sees a truncated incoming flow.
                conn.streamByIdLocked(frame.streamId)?.closeIncoming()
            }

            is StopSendingFrame -> {
-                // Audit-4 #2: peer asks us to stop sending on its read side.
+                // 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 peer using STOP_SENDING to back-pressure
-                // would have to fall back to MAX_STREAM_DATA = 0, which we
-                // already honour.
+                // and dropped. A future enhancement should emit RESET_STREAM
+                // back per RFC 9000 §3.5.
                ackEliciting = true
            }

            is NewTokenFrame -> {
-                // Audit-4 #2: 0-RTT/resumption token. Out-of-scope; drop.
+                // Round-4 #2: 0-RTT/resumption token. Out-of-scope; drop.
                ackEliciting = true
            }

            is NewConnectionIdFrame -> {
-                // We don't support migration; ignore.
+                // RFC 9000 §13.2.1: NEW_CONNECTION_ID is ack-eliciting. We
+                // don't support migration but still need to ACK to keep
+                // peer's loss-recovery happy.
+                ackEliciting = true
            }

            is ConnectionCloseFrame -> {
@@ -338,7 +353,15 @@ private fun dispatchFrames(
                    )
                    return
                }
-                conn.status = QuicConnection.Status.CONNECTED
+                ackEliciting = true
+                // Round-5 #13: only flip to CONNECTED if we're still in
+                // HANDSHAKING. Pre-fix this unconditionally overwrote the
+                // status, which would resurrect a connection that
+                // applyPeerTransportParameters had just closed via
+                // markClosedExternally because of a CID validation failure.
+                if (conn.status == QuicConnection.Status.HANDSHAKING) {
+                    conn.status = QuicConnection.Status.CONNECTED
+                }
            }

            is PingFrame -> {
@@ -329,12 +329,13 @@ private fun appendFlowControlUpdates(
 ) {
    val cfg = conn.config
    var totalRecvAdvanced = 0L
-    // Round-4 perf #9: only walk streams flagged by the parser since the last
+    // Round-4 perf #9 + round-5 #9: walk the streams via the index-friendly
+    // list view (no `entries.toList()` allocation), and only do per-stream
+    // window/threshold work for streams flagged by the parser since the last
    // drain. Streams whose receive frontier hasn't advanced cannot need a
-    // new MAX_STREAM_DATA frame, so iterating them is wasted work. The
-    // connection-level totalRecvAdvanced sum still requires looking at each
-    // stream's contiguousEnd, but only when the dirty flag is set.
-    for ((id, stream) in conn.streamsLocked()) {
+    // new MAX_STREAM_DATA frame.
+    for (stream in conn.streamsListLocked()) {
+        val id = stream.streamId
        val rcv = stream.receive.contiguousEnd()
        if (rcv == 0L) continue
        if (!stream.receiveDirtyForFlowControl) {
@@ -27,6 +27,7 @@ import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.launch

@@ -140,6 +141,14 @@ class QuicWebTransportSessionState(
     */
    val peerGoawayStreamId get() = demux.peerGoawayStreamId

+    /**
+     * Non-null when the peer's CONTROL stream produced a protocol error that
+     * the demux can't act on by itself (e.g. an H3_ID_ERROR GOAWAY id
+     * regression — round-5 #4). Applications should poll this and close the
+     * connection if set.
+     */
+    val peerGoawayProtocolError get() = demux.peerGoawayProtocolError
+
    /**
     * The WT_CLOSE_SESSION capsule the peer sent on the CONNECT bidi, or null
     * if no graceful close has arrived yet. Applications wanting to react
@@ -205,5 +214,20 @@ class QuicWebTransportSessionState(
            it.send.finish()
        }
        driver.close()
+        // Round-5 concurrency #1: cancel the WT scope so the demux pump
+        // and capsule reader coroutines launched in init{} actually exit.
+        // Pre-fix they kept running past close, holding references to
+        // the QuicStream / chunk channels indefinitely and producing
+        // memory growth on long sessions that opened/closed many WT
+        // sessions.
+        scope.cancel()
+        // Round-5 #8: if any caller is suspended on awaitPeerClose() and
+        // we're tearing down without ever observing a peer-initiated
+        // close, fail the deferred so the awaiter exits.
+        if (!peerCloseDeferred.isCompleted) {
+            peerCloseDeferred.cancel(
+                kotlinx.coroutines.CancellationException("WebTransport session closed locally"),
+            )
+        }
    }
 }
@@ -79,6 +79,18 @@ class WtPeerStreamDemux(
    var peerGoawayStreamId: Long? = null
        private set

+    /**
+     * Round-5 #4: surface H3_ID_ERROR (RFC 9114 §5.2 violation: GOAWAY id
+     * increased) so the QUIC layer can act on it instead of having the
+     * route()-level `catch (_: Throwable)` swallow it. Stays null until a
+     * regressing GOAWAY arrives. Application code (or
+     * [QuicWebTransportSessionState]) should poll this and close the
+     * connection if non-null.
+     */
+    @Volatile
+    var peerGoawayProtocolError: String? = null
+        private set
+
    val incomingStrippedStreams: Flow<StrippedWtStream> = readyStreams.consumeAsFlow()

    /**
@@ -90,85 +102,100 @@ class WtPeerStreamDemux(
    }

    private suspend fun route(stream: QuicStream) {
-        // Build a buffered chunk source: keeps unconsumed bytes until we
-        // know what kind of stream we're looking at.
-        val pending = ArrayDeque<ByteArray>()
-        val flowIterator = stream.incoming
-        val chunkChannel = Channel<ByteArray>(Channel.UNLIMITED)
-        scope.launch {
+        // Round-5 concurrency #2: wrap the whole route in coroutineScope so
+        // the inner collector launched below is joined on EVERY exit path,
+        // not just the ones that called drainBlackHole. Pre-fix four early-
+        // return sites (mismatched stream-type prefixes, unknown WT signal,
+        // foreign session id) left the collector orphaned, draining
+        // stream.incoming into a chunkChannel nobody read — unbounded
+        // memory growth per misbehaving peer stream.
+        kotlinx.coroutines.coroutineScope {
+            val pending = ArrayDeque<ByteArray>()
+            val flowIterator = stream.incoming
+            val chunkChannel = Channel<ByteArray>(Channel.UNLIMITED)
+            val collector =
+                launch {
+                    try {
+                        flowIterator.collect { chunkChannel.send(it) }
+                    } finally {
+                        chunkChannel.close()
+                    }
+                }
+
+            // Helper: read the next available bytes; returns null on stream close
+            // before enough bytes are present.
+            suspend fun moreBytes(): Boolean {
+                val chunk = chunkChannel.receiveCatching().getOrNull() ?: return false
+                pending.addLast(chunk)
+                return true
+            }
+
+            suspend fun readVarintFromPending(): Long? {
+                while (true) {
+                    val flat = flatten(pending)
+                    val res = Varint.decode(flat, 0)
+                    if (res != null) {
+                        consumeFromPending(pending, res.bytesConsumed)
+                        return res.value
+                    }
+                    if (!moreBytes()) return null
+                }
+            }
+
            try {
-                flowIterator.collect { chunkChannel.send(it) }
-            } finally {
-                chunkChannel.close()
-            }
-        }
-
-        // Helper: read the next available bytes; returns null on stream close
-        // before enough bytes are present.
-        suspend fun moreBytes(): Boolean {
-            val chunk = chunkChannel.receiveCatching().getOrNull() ?: return false
-            pending.addLast(chunk)
-            return true
-        }
-
-        suspend fun readVarintFromPending(): Long? {
-            while (true) {
-                val flat = flatten(pending)
-                val res = Varint.decode(flat, 0)
-                if (res != null) {
-                    consumeFromPending(pending, res.bytesConsumed)
-                    return res.value
-                }
-                if (!moreBytes()) return null
-            }
-        }
-
-        try {
-            if (StreamId.isUnidirectional(stream.streamId)) {
-                val streamType = readVarintFromPending() ?: return
-                when (streamType) {
-                    Http3StreamType.CONTROL -> {
-                        drainControlStream(pending, chunkChannel)
+                if (StreamId.isUnidirectional(stream.streamId)) {
+                    val streamType = readVarintFromPending()
+                    if (streamType == null) {
+                        collector.cancel()
+                        return@coroutineScope
                    }
-
-                    Http3StreamType.QPACK_ENCODER, Http3StreamType.QPACK_DECODER -> {
-                        drainBlackHole(chunkChannel)
-                    }
-
-                    Http3StreamType.WEBTRANSPORT_UNI_STREAM -> {
-                        val quarter = readVarintFromPending() ?: return
-                        if (quarter * 4L != expectedConnectStreamId) {
-                            drainBlackHole(chunkChannel) // not our session
-                            return
+                    when (streamType) {
+                        Http3StreamType.CONTROL -> {
+                            drainControlStream(pending, chunkChannel)
                        }
-                        emitStripped(stream, pending, chunkChannel, isUni = true)
-                    }

-                    else -> {
+                        Http3StreamType.QPACK_ENCODER, Http3StreamType.QPACK_DECODER -> {
+                            drainBlackHole(chunkChannel)
+                        }
+
+                        Http3StreamType.WEBTRANSPORT_UNI_STREAM -> {
+                            val quarter = readVarintFromPending()
+                            if (quarter == null || quarter * 4L != expectedConnectStreamId) {
+                                drainBlackHole(chunkChannel) // not our session / truncated
+                                return@coroutineScope
+                            }
+                            emitStripped(stream, pending, chunkChannel, isUni = true)
+                        }
+
+                        else -> {
+                            drainBlackHole(chunkChannel) // unknown — drop per RFC 9114 §9
+                        }
+                    }
+                } else {
+                    // Server-initiated bidi: per draft-ietf-webtrans-http3, prefixed
+                    // with WT_BIDI_STREAM (0x41) varint then quarter session id.
+                    val signal = readVarintFromPending()
+                    if (signal == null || signal != WtStreamType.WT_BIDI_STREAM) {
                        drainBlackHole(chunkChannel)
-                    } // unknown — drop per RFC 9114 §9
+                        return@coroutineScope
+                    }
+                    val quarter = readVarintFromPending()
+                    if (quarter == null || quarter * 4L != expectedConnectStreamId) {
+                        drainBlackHole(chunkChannel)
+                        return@coroutineScope
+                    }
+                    emitStripped(stream, pending, chunkChannel, isUni = false)
                }
-            } else {
-                // Server-initiated bidi: per draft-ietf-webtrans-http3, prefixed
-                // with WT_BIDI_STREAM (0x41) varint then quarter session id.
-                val signal = readVarintFromPending() ?: return
-                if (signal != WtStreamType.WT_BIDI_STREAM) {
-                    drainBlackHole(chunkChannel)
-                    return
-                }
-                val quarter = readVarintFromPending() ?: return
-                if (quarter * 4L != expectedConnectStreamId) {
-                    drainBlackHole(chunkChannel)
-                    return
-                }
-                emitStripped(stream, pending, chunkChannel, isUni = false)
+            } catch (ce: kotlinx.coroutines.CancellationException) {
+                // Audit-4 #17: don't swallow cancellation — needs to propagate
+                // to actually tear down the coroutine when scope is cancelled.
+                throw ce
+            } catch (_: Throwable) {
+                // peer closed mid-prefix or framing error — drop quietly. The
+                // surrounding coroutineScope joins the collector on exit, so
+                // no leak even on swallowed errors.
+                collector.cancel()
            }
-        } catch (ce: kotlinx.coroutines.CancellationException) {
-            // Audit-4 #17: don't swallow cancellation — needs to propagate
-            // to actually tear down the coroutine when scope is cancelled.
-            throw ce
-        } catch (_: Throwable) {
-            // peer closed mid-prefix or framing error — drop quietly
        }
    }

@@ -209,8 +236,16 @@ class WtPeerStreamDemux(
                    if (res != null) {
                        val prev = peerGoawayStreamId
                        if (prev != null && res.value > prev) {
+                            // Round-5 #4: surface the protocol error via
+                            // peerGoawayProtocolError so the QUIC layer can
+                            // close the connection. Throwing also exits this
+                            // CONTROL-stream reader; the surrounding route()
+                            // catch handles cleanup (collector cancel +
+                            // chunkChannel close).
+                            peerGoawayProtocolError =
+                                "H3_ID_ERROR: GOAWAY id increased ($prev → ${res.value})"
                            throw com.vitorpamplona.quic.QuicCodecException(
-                                "H3_ID_ERROR: GOAWAY id increased ($prev → ${res.value})",
+                                peerGoawayProtocolError!!,
                            )
                        }
                        peerGoawayStreamId = res.value
@@ -0,0 +1,160 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quic.connection
+
+import com.vitorpamplona.quic.frame.HandshakeDoneFrame
+import com.vitorpamplona.quic.frame.MaxDataFrame
+import com.vitorpamplona.quic.frame.MaxStreamDataFrame
+import com.vitorpamplona.quic.frame.MaxStreamsFrame
+import com.vitorpamplona.quic.frame.PaddingFrame
+import com.vitorpamplona.quic.frame.ResetStreamFrame
+import com.vitorpamplona.quic.tls.PermissiveCertificateValidator
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+
+/**
+ * Round-5 regression: every frame the parser dispatches MUST set
+ * `ackEliciting = true` if RFC 9000 §13.2.1 lists the frame type as
+ * ack-eliciting. Pre-fix the new ACK gating (round-4 perf #1) caused a
+ * packet carrying only e.g. MAX_DATA or HANDSHAKE_DONE to never trigger an
+ * ACK — the peer would PTO-retransmit forever.
+ *
+ * Tests drive each frame through a CONNECTED client and assert that a
+ * subsequent drainOutbound produces a packet (which contains the ACK).
+ */
+class AckElicitingFramesTest {
+    private fun connectedClient(): Pair<QuicConnection, InMemoryQuicPipe> {
+        val client =
+            QuicConnection(
+                serverName = "example.test",
+                config = QuicConnectionConfig(),
+                tlsCertificateValidator = PermissiveCertificateValidator(),
+            )
+        val pipe = InMemoryQuicPipe(client, client.destinationConnectionId.bytes)
+        client.start()
+        pipe.drive(maxRounds = 16)
+        check(client.status == QuicConnection.Status.CONNECTED)
+        // Drain any handshake-induced ACKs out of the way so subsequent
+        // tests see a clean state.
+        kotlinx.coroutines.runBlocking {
+            // Pull whatever the client has queued post-handshake; we don't
+            // care about the contents, only that the slate is clean.
+            while (drainOutbound(client, nowMillis = 0L) != null) { /* drain */ }
+        }
+        return client to pipe
+    }
+
+    @Test
+    fun max_data_alone_triggers_an_ack() {
+        // Pre-round-5: the parser handled MaxDataFrame without setting
+        // ackEliciting, so the round-4 ACK gate refused to emit an ACK.
+        val (client, pipe) = connectedClient()
+        // Server sends a packet carrying ONLY MAX_DATA (well, plus padding to
+        // hit the HP-sample minimum).
+        val packet =
+            pipe.buildServerApplicationDatagram(
+                listOf(MaxDataFrame(2_000_000), PaddingFrame, PaddingFrame, PaddingFrame),
+            )!!
+        feedDatagram(client, packet, nowMillis = 0L)
+        // The client should now want to send an ACK.
+        val out = drainOutbound(client, nowMillis = 0L)
+        assertNotNull(
+            out,
+            "MAX_DATA must be ack-eliciting; client must produce an ACK packet",
+        )
+    }
+
+    @Test
+    fun max_stream_data_alone_triggers_an_ack() {
+        val (client, pipe) = connectedClient()
+        // First open a stream so MaxStreamDataFrame has a target to update.
+        val packet =
+            pipe.buildServerApplicationDatagram(
+                listOf(
+                    MaxStreamDataFrame(streamId = 0L, maxStreamData = 50_000),
+                    PaddingFrame,
+                    PaddingFrame,
+                    PaddingFrame,
+                ),
+            )!!
+        feedDatagram(client, packet, nowMillis = 0L)
+        assertNotNull(drainOutbound(client, nowMillis = 0L))
+    }
+
+    @Test
+    fun max_streams_alone_triggers_an_ack() {
+        val (client, pipe) = connectedClient()
+        val packet =
+            pipe.buildServerApplicationDatagram(
+                listOf(
+                    MaxStreamsFrame(bidi = true, maxStreams = 100),
+                    PaddingFrame,
+                    PaddingFrame,
+                    PaddingFrame,
+                ),
+            )!!
+        feedDatagram(client, packet, nowMillis = 0L)
+        assertNotNull(drainOutbound(client, nowMillis = 0L))
+    }
+
+    @Test
+    fun handshake_done_alone_triggers_an_ack() {
+        val (client, pipe) = connectedClient()
+        // Pad heavily so HP-sample minimum is met.
+        val pings = List(40) { PaddingFrame }
+        val packet = pipe.buildServerApplicationDatagram(listOf(HandshakeDoneFrame()) + pings)!!
+        feedDatagram(client, packet, nowMillis = 0L)
+        assertNotNull(drainOutbound(client, nowMillis = 0L))
+    }
+
+    @Test
+    fun reset_stream_alone_triggers_an_ack() {
+        val (client, pipe) = connectedClient()
+        val frame = ResetStreamFrame(streamId = 1L, applicationErrorCode = 0L, finalSize = 0L)
+        val pings = List(40) { PaddingFrame }
+        val packet = pipe.buildServerApplicationDatagram(listOf(frame) + pings)!!
+        feedDatagram(client, packet, nowMillis = 0L)
+        assertNotNull(drainOutbound(client, nowMillis = 0L))
+    }
+
+    @Test
+    fun reset_stream_on_client_uni_id_closes_connection() {
+        // Round-5 #2: peer can't RESET_STREAM a stream we own the only side
+        // of (CLIENT_UNI). It's STREAM_STATE_ERROR.
+        val (client, pipe) = connectedClient()
+        val clientUniId = 2L // id % 4 == 2 → CLIENT_UNI
+        val frame =
+            ResetStreamFrame(
+                streamId = clientUniId,
+                applicationErrorCode = 0L,
+                finalSize = 0L,
+            )
+        val pings = List(40) { PaddingFrame }
+        val packet = pipe.buildServerApplicationDatagram(listOf(frame) + pings)!!
+        feedDatagram(client, packet, nowMillis = 0L)
+        assertEquals(
+            QuicConnection.Status.CLOSED,
+            client.status,
+            "RESET_STREAM on CLIENT_UNI is STREAM_STATE_ERROR; peer has no send side",
+        )
+    }
+}
@@ -53,11 +53,19 @@ class JdkCertificateValidator(
        expectedHost: String,
    ) {
        if (chain.isEmpty()) throw QuicCodecException("server sent empty certificate chain")
+        // Round-5 #5: parse certificates inside the try block so a malformed
+        // DER blob throws QuicCodecException (which the read loop maps to
+        // CONNECTION_CLOSE) instead of an uncaught CertificateException.
        val cf = CertificateFactory.getInstance("X.509")
-        val parsed =
-            chain.map {
-                cf.generateCertificate(ByteArrayInputStream(it)) as X509Certificate
-            }
+        val parsed: List<X509Certificate>
+        try {
+            parsed =
+                chain.map {
+                    cf.generateCertificate(ByteArrayInputStream(it)) as X509Certificate
+                }
+        } catch (t: Throwable) {
+            throw QuicCodecException("certificate chain parse failed: ${t.message}", t)
+        }
        try {
            // X509TrustManager auth-type string is the TLS key-exchange / sig-alg
            // pair derived from the cipher suite name — for TLS 1.3 we use the
@@ -187,16 +195,28 @@ class JdkCertificateValidator(
        }

    /**
-     * Pattern-match check for IPv4 / IPv6 literals so we don't trigger DNS
-     * lookups for hostnames during cert validation (audit-4 #4). IPv4 is
-     * "digits and dots only"; IPv6 is "contains a colon". Bracketed IPv6
-     * literals (`[::1]`) are accepted by stripping the brackets first.
+     * Strict pattern-match for IPv4 / IPv6 literals — round-5 #3 tightens
+     * audit-4 #4. The previous check (`all digits/dots and contains a dot`)
+     * accepted strings like "1.2.3.4.5" or "1.2" that Java's
+     * `InetAddress.getByName` happily resolves via DNS, defeating the SNI-
+     * leak fix. IPv4 must be exactly four dot-separated octets each in
+     * 0..255; IPv6 must contain a colon (further parsing is left to the
+     * JDK once we've confirmed it's a literal).
     */
    private fun looksLikeIpLiteral(host: String): Boolean {
        val unbracketed =
            if (host.startsWith("[") && host.endsWith("]")) host.substring(1, host.length - 1) else host
-        if (unbracketed.contains(':')) return true // IPv6
-        return unbracketed.all { it.isDigit() || it == '.' } && unbracketed.contains('.')
+        if (unbracketed.contains(':')) return true // IPv6 — parse via JDK
+        // IPv4: 4 octets 0..255, no leading zeros tolerated as integers.
+        val parts = unbracketed.split('.')
+        if (parts.size != 4) return false
+        for (p in parts) {
+            if (p.isEmpty() || p.length > 3) return false
+            if (!p.all { it.isDigit() }) return false
+            val n = p.toIntOrNull() ?: return false
+            if (n !in 0..255) return false
+        }
+        return true
    }

    private fun dnsMatches(
@@ -0,0 +1,128 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quic.tls
+
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+/**
+ * Round-5 #3: `looksLikeIpLiteral` must NOT accept ambiguous strings that
+ * Java's InetAddress.getByName would resolve via DNS. Pre-fix
+ * "all digits and dots" + "contains a dot" let through "1.2.3.4.5", "1.2",
+ * and "1." — all of which Java happily resolves as DNS hostnames, leaking
+ * the SNI/hostname over plaintext DNS during cert validation.
+ *
+ * The function is private; tests reach it via [JdkCertificateValidator]'s
+ * `hostnameMatches` indirectly, but more reliably we expose a direct unit
+ * test by reflection on the private method.
+ */
+class JdkCertificateValidatorIpLiteralTest {
+    private val validator = JdkCertificateValidator()
+
+    private fun looksLikeIpLiteral(host: String): Boolean {
+        val method =
+            JdkCertificateValidator::class.java
+                .getDeclaredMethod("looksLikeIpLiteral", String::class.java)
+                .apply { isAccessible = true }
+        return method.invoke(validator, host) as Boolean
+    }
+
+    @Test
+    fun standard_ipv4_addresses_match() {
+        assertTrue(looksLikeIpLiteral("127.0.0.1"))
+        assertTrue(looksLikeIpLiteral("0.0.0.0"))
+        assertTrue(looksLikeIpLiteral("255.255.255.255"))
+        assertTrue(looksLikeIpLiteral("192.168.1.1"))
+        assertTrue(looksLikeIpLiteral("10.0.0.42"))
+    }
+
+    @Test
+    fun ipv6_literals_match() {
+        // Bracketed and unbracketed forms.
+        assertTrue(looksLikeIpLiteral("::1"))
+        assertTrue(looksLikeIpLiteral("[::1]"))
+        assertTrue(looksLikeIpLiteral("2001:db8::1"))
+        assertTrue(looksLikeIpLiteral("[2001:db8::1]"))
+        assertTrue(looksLikeIpLiteral("fe80::1"))
+    }
+
+    @Test
+    fun more_than_four_octets_does_not_match() {
+        // Pre-fix: "all digits and dots, contains a dot" → true.
+        // Post-fix: must require exactly 4 octets.
+        assertFalse(
+            looksLikeIpLiteral("1.2.3.4.5"),
+            "5-octet string is not a valid IPv4 literal",
+        )
+    }
+
+    @Test
+    fun fewer_than_four_octets_does_not_match() {
+        assertFalse(looksLikeIpLiteral("1.2"))
+        assertFalse(looksLikeIpLiteral("1.2.3"))
+        assertFalse(looksLikeIpLiteral("1."))
+        assertFalse(looksLikeIpLiteral("."))
+        assertFalse(looksLikeIpLiteral(""))
+    }
+
+    @Test
+    fun octets_above_255_do_not_match() {
+        assertFalse(looksLikeIpLiteral("256.0.0.1"))
+        assertFalse(looksLikeIpLiteral("999.999.999.999"))
+        assertFalse(looksLikeIpLiteral("1.2.3.300"))
+    }
+
+    @Test
+    fun hostnames_do_not_match() {
+        // The whole point: hostnames must be rejected so we don't trigger
+        // a DNS lookup.
+        assertFalse(looksLikeIpLiteral("example.com"))
+        assertFalse(looksLikeIpLiteral("127-0-0-1.example.com"))
+        assertFalse(looksLikeIpLiteral("nests.io"))
+        assertFalse(looksLikeIpLiteral("a.b.c.d.e"))
+    }
+
+    @Test
+    fun empty_octet_does_not_match() {
+        assertFalse(looksLikeIpLiteral("1..2.3"))
+        assertFalse(looksLikeIpLiteral(".1.2.3.4"))
+        assertFalse(looksLikeIpLiteral("1.2.3.4."))
+    }
+
+    @Test
+    fun non_digit_characters_in_octet_do_not_match() {
+        assertFalse(looksLikeIpLiteral("1.2.3.x"))
+        assertFalse(looksLikeIpLiteral("a.b.c.d"))
+    }
+
+    // The exact contract under audit-4 #4: tightened so non-IP-literal
+    // strings never trigger DNS during cert validation. Quick smoke.
+    @Test
+    fun count_summary() {
+        // The pre-fix accepted; the post-fix rejects.
+        val rejected =
+            listOf("1.2.3.4.5", "1.2", "1.", ".", "", "256.0.0.1", "1..2.3", "example.com")
+                .count { !looksLikeIpLiteral(it) }
+        assertEquals(8, rejected, "all 8 ambiguous strings must be rejected by the tightened pattern")
+    }
+}