test(quic): regression tests for receive-limit, incoming channel cap, coalesced-packet skip

Three pending audit-2/3 regression tests, plus the InMemoryQuicPipe helpers
needed to drive them.

ReceiveLimitEnforcementTest — peer overshooting per-stream receive limit
must transition the connection to CLOSED via markClosedExternally. Mirror
test verifies the boundary value (frameEnd == receiveLimit) does NOT close.

QuicStreamIncomingChannelTest — the per-stream incoming channel is bounded
at 64 chunks; trySend on saturation must not block (would deadlock the
parser on the connection lock). Empty chunks are filtered. closeIncoming
terminates the collector.

CoalescedPacketSkipTest — RFC 9000 §12.2 / RFC 9001 §5.5: feedDatagram
must walk across coalesced packets, must skip a packet that fails AEAD
verification using peekHeader.totalLength (not break the loop), and must
exit cleanly when a trailing header is truncated.

Pipe additions: buildServerApplicationDatagram + coalesceDatagrams give
tests the primitives to drive arbitrary server → client app-level frames.
InMemoryQuicPipe also takes an optional tlsServer so tests can advertise
non-default transport parameters.

https://claude.ai/code/session_01EC1tfXfap8k8GyKvrxkxZx

quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CoalescedPacketSkipTest.kt

@@ -0,0 +1,175 @@
+/*
+ * 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.crypto.Aes128Gcm
+import com.vitorpamplona.quic.crypto.AesEcbHeaderProtection
+import com.vitorpamplona.quic.crypto.InitialProtection
+import com.vitorpamplona.quic.crypto.InitialSecrets
+import com.vitorpamplona.quic.crypto.PlatformAesOneBlock
+import com.vitorpamplona.quic.frame.PingFrame
+import com.vitorpamplona.quic.frame.encodeFrames
+import com.vitorpamplona.quic.packet.LongHeaderPacket
+import com.vitorpamplona.quic.packet.LongHeaderPlaintextPacket
+import com.vitorpamplona.quic.packet.LongHeaderType
+import com.vitorpamplona.quic.packet.QuicVersion
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+/**
+ * Coalesced-packet skip behaviour, RFC 9000 §12.2 + RFC 9001 §5.5.
+ *
+ * Multiple QUIC packets can share a single UDP datagram. The receive loop in
+ * [feedDatagram] walks across them by trusting the long-header `length`
+ * field. Two cases the audit-3 review specifically called out:
+ *
+ *   1. Two valid packets coalesced — both must be observed (the loop must
+ *      not stop at the first).
+ *   2. A packet whose AEAD verification fails — the receiver must drop ONLY
+ *      that packet, advance using `peekHeader`'s totalLength, and continue
+ *      with the next one (RFC 9001 §5.5: "the receiver MUST attempt to
+ *      process all coalesced packets").
+ *
+ * Pre-fix the loop broke on first decrypt failure, dropping any subsequent
+ * packets in the datagram on the floor — a silent data-loss bug that would
+ * have surfaced as flaky handshakes.
+ */
+class CoalescedPacketSkipTest {
+    private val hp = AesEcbHeaderProtection(PlatformAesOneBlock)
+
+    private fun buildInitial(
+        client: QuicConnection,
+        secrets: InitialProtection,
+        packetNumber: Long,
+        scid: ConnectionId,
+        payload: ByteArray,
+    ): ByteArray =
+        LongHeaderPacket.build(
+            LongHeaderPlaintextPacket(
+                type = LongHeaderType.INITIAL,
+                version = QuicVersion.V1,
+                dcid = client.sourceConnectionId,
+                scid = scid,
+                packetNumber = packetNumber,
+                payload = payload,
+            ),
+            Aes128Gcm,
+            secrets.serverKey,
+            secrets.serverIv,
+            hp,
+            secrets.serverHp,
+            largestAckedInSpace = -1L,
+        )
+
+    @Test
+    fun two_coalesced_initial_packets_are_both_observed() {
+        // Fresh client: Initial keys auto-install in the constructor based on
+        // its DCID. We don't drive the handshake — we just want to verify the
+        // datagram-level loop walks across both packets.
+        val client =
+            QuicConnection(
+                serverName = "example.test",
+                config = QuicConnectionConfig(),
+                tlsCertificateValidator = null,
+            )
+        val secrets = InitialSecrets.derive(client.destinationConnectionId.bytes)
+        val serverScid = ConnectionId.random(8)
+
+        // PING + PADDING. The padding is required because header protection
+        // samples 16 bytes starting 4 bytes past the PN offset — a bare PING
+        // (1 byte plaintext + 16 byte AEAD tag) doesn't leave enough.
+        val ping = encodeFrames(listOf(PingFrame)) + ByteArray(24)
+        val pkt0 = buildInitial(client, secrets, packetNumber = 0L, scid = serverScid, payload = ping)
+        val pkt1 = buildInitial(client, secrets, packetNumber = 1L, scid = serverScid, payload = ping)
+
+        // Concatenate into one datagram (RFC 9000 §12.2 coalesced shape).
+        val coalesced = pkt0 + pkt1
+        feedDatagram(client, coalesced, nowMillis = 0L)
+
+        // If the loop stopped after the first packet we'd have largestReceived=0.
+        // Both packets observed → largestReceived advances to 1.
+        assertEquals(
+            1L,
+            client.initial.pnSpace.largestReceived,
+            "second coalesced packet must also be processed; loop must walk past the first",
+        )
+    }
+
+    @Test
+    fun corrupted_first_packet_does_not_swallow_a_valid_second() {
+        val client =
+            QuicConnection(
+                serverName = "example.test",
+                config = QuicConnectionConfig(),
+                tlsCertificateValidator = null,
+            )
+        val secrets = InitialSecrets.derive(client.destinationConnectionId.bytes)
+        val serverScid = ConnectionId.random(8)
+
+        val ping = encodeFrames(listOf(PingFrame)) + ByteArray(24)
+        val pkt0Good = buildInitial(client, secrets, packetNumber = 0L, scid = serverScid, payload = ping)
+        // Corrupt the AEAD tag (last byte) of the first packet — header still
+        // parses cleanly under peekHeader, but parseAndDecrypt fails GCM
+        // verification and returns null.
+        val pkt0Corrupt = pkt0Good.copyOf()
+        pkt0Corrupt[pkt0Corrupt.size - 1] = (pkt0Corrupt[pkt0Corrupt.size - 1].toInt() xor 0x01).toByte()
+
+        val pkt1 = buildInitial(client, secrets, packetNumber = 1L, scid = serverScid, payload = ping)
+
+        feedDatagram(client, pkt0Corrupt + pkt1, nowMillis = 0L)
+
+        // Pre-fix: loop broke on the decrypt failure, largestReceived stays -1.
+        // Post-fix: loop advances by peekHeader.totalLength and processes pkt1.
+        assertEquals(
+            1L,
+            client.initial.pnSpace.largestReceived,
+            "valid second packet must still be processed when the first fails decrypt",
+        )
+    }
+
+    @Test
+    fun feed_stops_cleanly_when_header_is_truncated_inside_a_coalesced_run() {
+        // Defensive case: if a coalesced run ends with a truncated header
+        // (peekHeader returns null), the loop must `break` rather than spin
+        // or read past the buffer. The first packet's effects MUST still be
+        // observable.
+        val client =
+            QuicConnection(
+                serverName = "example.test",
+                config = QuicConnectionConfig(),
+                tlsCertificateValidator = null,
+            )
+        val secrets = InitialSecrets.derive(client.destinationConnectionId.bytes)
+        val serverScid = ConnectionId.random(8)
+
+        val ping = encodeFrames(listOf(PingFrame)) + ByteArray(24)
+        val pkt0 = buildInitial(client, secrets, packetNumber = 0L, scid = serverScid, payload = ping)
+
+        // Append two bytes that look like the start of a long-header packet
+        // (high bit set) but aren't enough for peekHeader to succeed.
+        val truncated = pkt0 + byteArrayOf(0xC0.toByte(), 0x00)
+
+        feedDatagram(client, truncated, nowMillis = 0L)
+
+        // First packet processed, loop exits cleanly without throwing.
+        assertEquals(0L, client.initial.pnSpace.largestReceived)
+    }
+}

quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt

@@ -254,6 +254,58 @@ class InMemoryQuicPipe(
     private var initialCryptoOffset: Long = 0L
     private var handshakeCryptoOffset: Long = 0L
 
+    /**
+     * Build a 1-RTT (short-header) datagram from the server containing the
+     * given frames, encrypted with the negotiated server-side application
+     * keys. Tests use this to drive flow-control violations, MAX_STREAMS
+     * frames, and other peer-initiated frame paths against the real client.
+     *
+     * The handshake must have completed (server has 1-RTT keys) — calling
+     * before that returns null.
+     */
+    fun buildServerApplicationDatagram(frames: List<com.vitorpamplona.quic.frame.Frame>): ByteArray? {
+        val proto = serverApplicationTx ?: return null
+        val pn = applicationPnSpace.allocateOutbound()
+        val payload = encodeFrames(frames)
+        return ShortHeaderPacket.build(
+            com.vitorpamplona.quic.packet.ShortHeaderPlaintextPacket(
+                dcid = client.sourceConnectionId,
+                packetNumber = pn,
+                payload = payload,
+            ),
+            proto.aead,
+            proto.key,
+            proto.iv,
+            proto.hp,
+            proto.hpKey,
+            largestAckedInSpace = applicationPnSpace.largestReceived,
+        )
+    }
+
+    /**
+     * Concatenate multiple already-built packets into a single UDP datagram —
+     * the wire-level shape of QUIC coalesced packets (RFC 9000 §12.2). Tests
+     * use this to verify the client's loop in `feedDatagram` correctly walks
+     * across packet boundaries instead of stopping at the first packet.
+     */
+    fun coalesceDatagrams(packets: List<ByteArray>): ByteArray {
+        var total = 0
+        for (p in packets) total += p.size
+        val out = ByteArray(total)
+        var pos = 0
+        for (p in packets) {
+            p.copyInto(out, pos)
+            pos += p.size
+        }
+        return out
+    }
+
+    /**
+     * Build a 1-RTT packet without coalescing it — used together with
+     * [coalesceDatagrams] to construct hostile/multi-packet datagrams.
+     */
+    fun buildServerApplicationPacket(frames: List<com.vitorpamplona.quic.frame.Frame>): ByteArray? = buildServerApplicationDatagram(frames)
+
     private fun buildServerInitialPacket(crypto: ByteArray): ByteArray {
         val proto =
             PacketProtection(

quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ReceiveLimitEnforcementTest.kt

@@ -0,0 +1,149 @@
+/*
+ * 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.StreamFrame
+import com.vitorpamplona.quic.tls.InProcessTlsServer
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotEquals
+
+/**
+ * Verifies the receive-side flow-control enforcement landed by audit-2.
+ *
+ * RFC 9000 §4.1: a peer that sends bytes past the receiver's advertised
+ * MAX_STREAM_DATA limit MUST be torn down with FLOW_CONTROL_ERROR. Without
+ * this enforcement a hostile peer could pin arbitrary memory by streaming
+ * beyond the limit — the per-stream `incomingChannel` capacity (64 chunks)
+ * caps the immediate damage, but the connection-level kill is what stops a
+ * sustained attack.
+ *
+ * The parser implementation (QuicConnectionParser.kt:184) calls
+ * `markClosedExternally` when `frame.offset + frame.data.size >
+ * stream.receiveLimit`. These tests drive that path end-to-end via
+ * [InMemoryQuicPipe], using a TLS server that advertises tiny per-stream
+ * receive limits so we can produce the violation without huge buffers.
+ */
+class ReceiveLimitEnforcementTest {
+    @Test
+    fun peer_exceeding_per_stream_receive_limit_closes_connection() {
+        val client =
+            QuicConnection(
+                serverName = "example.test",
+                config =
+                    QuicConnectionConfig(
+                        // We advertise a small per-peer-bidi receive limit so the
+                        // server only needs to push 100 bytes to overflow. The
+                        // parser's check is `frameEnd > receiveLimit`, where
+                        // receiveLimit defaults to initialMaxStreamDataBidiRemote
+                        // for peer-initiated bidi streams.
+                        initialMaxStreamDataBidiRemote = 32,
+                        initialMaxStreamDataBidiLocal = 32,
+                    ),
+                tlsCertificateValidator = null,
+            )
+        val tlsServer =
+            InProcessTlsServer(
+                transportParameters =
+                    TransportParameters(
+                        initialMaxData = 1_000_000,
+                        initialMaxStreamDataBidiLocal = 100_000,
+                        initialMaxStreamDataBidiRemote = 100_000,
+                        initialMaxStreamDataUni = 100_000,
+                        initialMaxStreamsBidi = 16,
+                        initialMaxStreamsUni = 16,
+                    ).encode(),
+            )
+        val pipe = InMemoryQuicPipe(client, client.destinationConnectionId.bytes, tlsServer)
+        client.start()
+        pipe.drive(maxRounds = 16)
+        assertEquals(QuicConnection.Status.CONNECTED, client.status)
+
+        // Send a STREAM frame from server on a server-initiated bidi (id 1)
+        // with 64 bytes — twice the client's advertised 32-byte cap. The
+        // parser's frame loop must call markClosedExternally because the
+        // last byte (offset 0 + size 64 = 64) exceeds receiveLimit (32).
+        val streamId = 1L // server-initiated bidi: id % 4 == 1
+        val payload = ByteArray(64) { it.toByte() }
+        val streamFrame = StreamFrame(streamId = streamId, offset = 0L, data = payload, fin = false)
+        val packet = pipe.buildServerApplicationDatagram(listOf(streamFrame))
+        assertNotEquals(null, packet, "server must have application keys after handshake")
+        feedDatagram(client, packet!!, nowMillis = 0L)
+
+        // The connection is now CLOSED with the receive-limit reason. Without
+        // the audit-2 fix this would have stayed CONNECTED (silently buffering
+        // the over-limit bytes).
+        assertEquals(
+            QuicConnection.Status.CLOSED,
+            client.status,
+            "client must transition to CLOSED on receive-limit violation",
+        )
+    }
+
+    @Test
+    fun peer_within_per_stream_receive_limit_keeps_connection_open() {
+        // Mirror of the violation test: same setup, but the server stays
+        // strictly within the cap. The connection MUST remain CONNECTED, the
+        // bytes MUST land in the stream's incoming buffer. Catches a regression
+        // where the audit fix accidentally fires on the boundary value
+        // (`==` vs `>` confusion).
+        val client =
+            QuicConnection(
+                serverName = "example.test",
+                config =
+                    QuicConnectionConfig(
+                        initialMaxStreamDataBidiRemote = 64,
+                        initialMaxStreamDataBidiLocal = 64,
+                    ),
+                tlsCertificateValidator = null,
+            )
+        val tlsServer =
+            InProcessTlsServer(
+                transportParameters =
+                    TransportParameters(
+                        initialMaxData = 1_000_000,
+                        initialMaxStreamDataBidiLocal = 100_000,
+                        initialMaxStreamDataBidiRemote = 100_000,
+                        initialMaxStreamDataUni = 100_000,
+                        initialMaxStreamsBidi = 16,
+                        initialMaxStreamsUni = 16,
+                    ).encode(),
+            )
+        val pipe = InMemoryQuicPipe(client, client.destinationConnectionId.bytes, tlsServer)
+        client.start()
+        pipe.drive(maxRounds = 16)
+        assertEquals(QuicConnection.Status.CONNECTED, client.status)
+
+        // Send exactly 64 bytes — the cap. frameEnd == receiveLimit, which
+        // the parser permits (strict `>` check).
+        val streamId = 1L
+        val payload = ByteArray(64) { it.toByte() }
+        val frame = StreamFrame(streamId = streamId, offset = 0L, data = payload, fin = false)
+        val packet = pipe.buildServerApplicationDatagram(listOf(frame))!!
+        feedDatagram(client, packet, nowMillis = 0L)
+
+        assertEquals(
+            QuicConnection.Status.CONNECTED,
+            client.status,
+            "client at exact receive-limit boundary must remain CONNECTED",
+        )
+    }
+}

quic/src/commonTest/kotlin/com/vitorpamplona/quic/stream/QuicStreamIncomingChannelTest.kt

@@ -0,0 +1,142 @@
+/*
+ * 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.stream
+
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.test.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+
+/**
+ * Bounds-checking on the per-stream `incomingChannel`.
+ *
+ * Audit-2 finding: a slow consumer paired with an unbounded channel was a
+ * memory-pin vector — the parser would happily forward gigabytes of inbound
+ * STREAM bytes that nothing ever read. The fix capped the channel at 64
+ * chunks; the producer-side [QuicStream.deliverIncoming] uses `trySend`, so
+ * a saturated channel quietly drops the offending chunk.
+ *
+ * The defence-in-depth layer is the connection-level receive-limit check in
+ * the parser ([com.vitorpamplona.quic.connection.QuicConnectionParser] line
+ * ~184) which closes the connection before a hostile peer can stream past
+ * the advertised limit. These tests exercise the channel cap directly so a
+ * regression that bumps the capacity (or removes it) gets caught even if
+ * the receive-limit path moves.
+ */
+class QuicStreamIncomingChannelTest {
+    @Test
+    fun deliver_incoming_buffers_chunks_until_collector_drains_them() {
+        // Baseline: chunks delivered before any collector starts MUST still
+        // surface to the collector — the channel must buffer up to its cap.
+        runBlocking {
+            val stream = QuicStream(streamId = 0L, direction = QuicStream.Direction.BIDIRECTIONAL)
+
+            // Push three small chunks before any collector is around.
+            stream.deliverIncoming(byteArrayOf(0x10))
+            stream.deliverIncoming(byteArrayOf(0x20))
+            stream.deliverIncoming(byteArrayOf(0x30))
+            stream.closeIncoming()
+
+            val received = mutableListOf<ByteArray>()
+            stream.incoming.collect { received += it }
+
+            assertEquals(3, received.size)
+            assertContentEquals(byteArrayOf(0x10), received[0])
+            assertContentEquals(byteArrayOf(0x20), received[1])
+            assertContentEquals(byteArrayOf(0x30), received[2])
+        }
+    }
+
+    @Test
+    fun deliver_incoming_drops_chunks_past_channel_capacity_without_blocking() {
+        // The producer side uses trySend; once the channel is full and there
+        // is no consumer, additional chunks are silently dropped. The producer
+        // MUST NOT block — the parser is on the connection lock and a blocked
+        // producer would deadlock the whole connection.
+        runBlocking {
+            val stream = QuicStream(streamId = 0L, direction = QuicStream.Direction.BIDIRECTIONAL)
+            // Fill well past capacity (64). Each call must return immediately.
+            // Wrap in a withTimeoutOrNull guard: if the producer EVER blocks,
+            // this assertion fires.
+            val ok =
+                withTimeoutOrNull(2_000L) {
+                    repeat(1024) { i -> stream.deliverIncoming(byteArrayOf(i.toByte())) }
+                    true
+                }
+            assertEquals(
+                true,
+                ok,
+                "deliverIncoming must not block once the channel is saturated " +
+                    "— a blocked producer would deadlock the parser on the connection lock",
+            )
+
+            // We don't try to assert the exact dropped count — Channel's
+            // internal queue + buffer transitions make that brittle. We DO
+            // assert that AT MOST the channel capacity (64) chunks landed.
+            stream.closeIncoming()
+            val collected = mutableListOf<ByteArray>()
+            stream.incoming.collect { collected += it }
+            assert(collected.size <= 64) {
+                "channel surfaced ${collected.size} chunks; cap is 64"
+            }
+        }
+    }
+
+    @Test
+    fun deliver_incoming_skips_empty_chunks() {
+        // The parser issues a STREAM frame even when its data is zero-length
+        // (FIN-only frames). deliverIncoming MUST NOT push an empty chunk —
+        // otherwise the consumer sees a bogus "empty data event" between real
+        // bytes. The closeIncoming path is the FIN signal.
+        runBlocking {
+            val stream = QuicStream(streamId = 0L, direction = QuicStream.Direction.BIDIRECTIONAL)
+            stream.deliverIncoming(ByteArray(0))
+            stream.deliverIncoming(byteArrayOf(0x42))
+            stream.closeIncoming()
+
+            val first = withTimeoutOrNull(2_000L) { stream.incoming.first() }
+            assertNotNull(first)
+            assertContentEquals(byteArrayOf(0x42), first, "first emitted chunk must be the non-empty one")
+        }
+    }
+
+    @Test
+    fun close_incoming_terminates_collector_immediately_when_buffer_is_empty() {
+        runBlocking {
+            val stream = QuicStream(streamId = 0L, direction = QuicStream.Direction.BIDIRECTIONAL)
+            stream.closeIncoming()
+            val collected = mutableListOf<ByteArray>()
+            // collect must terminate, not hang. The withTimeoutOrNull guards
+            // a regression that would, e.g., switch from `consumeAsFlow` to
+            // a flow that never closes when the channel does.
+            val finished =
+                withTimeoutOrNull(2_000L) {
+                    stream.incoming.collect { collected += it }
+                    true
+                }
+            assertEquals(true, finished)
+            assertEquals(0, collected.size)
+        }
+    }
+}