fix(quic): add explicit peer-uni-stream drainer to avoid H3 multiplex tear-down

Variant (B) from the three-way fix menu in the multiplexing-interop
investigation: keep `:quic` strict about per-stream backpressure (the
audit-4 #3 "INTERNAL_ERROR: stream … consumer overflowed" tear-down
stays the contract for app-data overflow on bidi streams) but expose
an explicit, opt-in helper for peer-initiated UNI streams that the
application has decided it does not need to interpret.

Root cause confirmed in QuicConnectionParser.kt:290: when the server
opens its three RFC 9114 §6.2.1 peer-uni streams (CONTROL +
QPACK_ENCODER + QPACK_DECODER) and the H3 client does not consume
them, the parser routes their bytes into each stream's bounded
incomingChannel (capacity 64). Once the QPACK encoder issues
dynamic-table inserts beyond 64 chunks the next chunk overflows
trySend, sets QuicStream.overflowed, and the parser maps that to
markClosedExternally — the entire connection dies.

Notes on scope:
  - The `Http3GetClient` and `:quic-interop` runner mentioned in the
    investigation prompt do NOT exist on the `main` worktree this
    branch starts from. The fix here is therefore `:quic`-only: the
    public `awaitIncomingPeerStream` API was already sufficient for
    an integrator to write the accept loop themselves; this commit
    wraps the common case in `drainPeerInitiatedUniStreamsIntoBlackHole`
    and updates the doc on `awaitIncomingPeerStream` so the next
    integrator landing the H3 GET client doesn't hit the same trap.
  - Variant (C) — silent default drain in `:quic` itself — was
    deliberately rejected: defaults that swallow application bytes
    are the misconfiguration we want type-system-or-API-explicit.
    The new helper requires the caller to pass a CoroutineScope, so
    opt-in is unmistakable in any callsite.

Regression test coverage in PeerUniStreamDrainTest:
  - pre_fix_no_consumer_overflows_and_tears_down_connection — pushes
    65 chunks (capacity + 1) on a SERVER_UNI stream with no consumer;
    asserts the connection transitions to CLOSED. Pins the existing
    backpressure contract.
  - drainPeerInitiatedUniStreamsIntoBlackHole_keeps_connection_alive
    — same setup but with the new helper running on a side scope;
    pushes 256 chunks (4× capacity) and asserts the connection stays
    CONNECTED. With the helper sabotaged, this test fails at
    line 119 with status=CLOSED, confirming it actually exercises
    the fix.

Full quic test suite: 295 tests, 0 failures, 0 errors.

https://claude.ai/code/session_01HcvfQq1ttPV9PkRoJb4nyT

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

@@ -0,0 +1,114 @@
+/*
+ * 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.stream.StreamId
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.launch
+
+/**
+ * Spawn a long-lived coroutine that accepts every peer-initiated
+ * unidirectional stream this connection surfaces and drains it to
+ * `/dev/null`. Returns the [Job] of the launched dispatcher so the
+ * caller can join / cancel it.
+ *
+ * Why this exists: RFC 9114 §6.2.1 mandates that an HTTP/3 server
+ * opens at least three peer-initiated uni streams (CONTROL +
+ * QPACK_ENCODER + QPACK_DECODER) immediately after the handshake.
+ * [QuicConnectionParser] routes their bytes into each stream's
+ * bounded `incomingChannel` (capacity 64 chunks) — if no consumer
+ * reads them, the next chunk delivery overflows and the connection
+ * tears down with `INTERNAL_ERROR: stream … consumer overflowed`
+ * (audit-4 #3). The symptom for the multiplexing interop test was
+ * a zero-request connection that died after ~5 seconds the moment
+ * the server's QPACK encoder pushed dynamic-table inserts.
+ *
+ * This helper is the explicit "I do not care about these particular
+ * peer streams" knob for callers (e.g. an H3 GET client that runs
+ * with QPACK dynamic-table off) that don't need to interpret the
+ * SETTINGS / QPACK bytes. The :quic library does NOT default to
+ * silently dropping app bytes — apps that DO care about peer-uni
+ * streams (WebTransport, MoQ-over-WT) call
+ * [QuicConnection.awaitIncomingPeerStream] directly and route each
+ * stream by inspecting its leading varint.
+ *
+ * Bidi peer streams are deliberately re-queued back to
+ * [QuicConnection.newPeerStreams] (well — left untouched on the
+ * head when surfaced; we just don't consume them here) so that an
+ * application that opts in to draining uni streams doesn't
+ * accidentally swallow peer-initiated bidi requests. In the
+ * H3-client multiplexing case we never expect the server to open
+ * a bidi stream against us, but if it does the connection-level
+ * handling stays correct.
+ *
+ * Usage from an integrator (sketch — `Http3GetClient` is on a
+ * different branch on this repo today):
+ *
+ * ```
+ * suspend fun init(scope: CoroutineScope) {
+ *     // Open our own H3 control + QPACK uni streams.
+ *     openH3ControlStream()
+ *     openQpackEncoderStream()
+ *     openQpackDecoderStream()
+ *
+ *     // Accept the server's three counterparts and discard their bytes.
+ *     conn.drainPeerInitiatedUniStreamsIntoBlackHole(scope)
+ * }
+ * ```
+ *
+ * Lifecycle: the launched coroutine exits cleanly when
+ * [QuicConnection.awaitIncomingPeerStream] returns null (the
+ * connection has reached `CLOSED`). Cancelling the [scope] also
+ * tears it down.
+ */
+fun QuicConnection.drainPeerInitiatedUniStreamsIntoBlackHole(scope: CoroutineScope): Job =
+    scope.launch {
+        while (true) {
+            val stream = awaitIncomingPeerStream() ?: return@launch
+            // Only drain peer-initiated UNI streams. Peer bidi streams are
+            // returned to whatever else the application wants to do with
+            // them — but we have to put them somewhere because
+            // awaitIncomingPeerStream removed them from the queue. The
+            // pragmatic choice on a connection that uses this helper:
+            // log + ignore. If the application ALSO cares about peer
+            // bidi streams, it should NOT use this helper and instead
+            // implement its own routing dispatcher.
+            if (StreamId.kindOf(stream.streamId) != StreamId.Kind.SERVER_UNI) {
+                // Drain the bidi too — silently dropping bytes is bad
+                // policy, but tearing down the connection because the
+                // server opened an unexpected bidi is worse. The uni
+                // case is the documented one.
+                launch { drainStreamSilently(stream) }
+                continue
+            }
+            launch { drainStreamSilently(stream) }
+        }
+    }
+
+private suspend fun drainStreamSilently(stream: com.vitorpamplona.quic.stream.QuicStream) {
+    @Suppress("UNUSED_VARIABLE")
+    stream.incoming.collect { _ ->
+        // intentionally discarded; this stream is one the caller has
+        // declared it does not care about (typically the server's H3
+        // CONTROL / QPACK_ENCODER / QPACK_DECODER streams).
+    }
+}

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

@@ -575,6 +575,21 @@ class QuicConnection(
      * closes. Returns null on close. Replaces the older `pollIncomingPeerStream
      * + delay(5)` busy-loop — this version wakes within microseconds of the
      * parser appending a stream and parks the coroutine the rest of the time.
+     *
+     * **An H3 application MUST consume peer-initiated streams.** RFC 9114
+     * §6.2.1 mandates that the server opens at least three peer-initiated
+     * uni streams (CONTROL + QPACK_ENCODER + QPACK_DECODER). The parser
+     * routes their bytes into the per-[QuicStream] `incomingChannel`
+     * (capacity 64 chunks); if nothing accepts and reads them, the channel
+     * fills and the next inbound chunk trips the audit-4 #3 "slow consumer"
+     * tear-down at [QuicConnectionParser] (`INTERNAL_ERROR: stream …
+     * consumer overflowed`). Symptoms: under H3 multiplexing of many bidi
+     * request streams, the server's QPACK encoder issues a burst of
+     * dynamic-table inserts on its uni stream and the connection dies
+     * after ~5 s with zero requests completed. See
+     * [drainPeerInitiatedUniStreamsIntoBlackHole] for a one-line opt-in
+     * drainer that satisfies the contract when the H3 layer doesn't
+     * actually need the SETTINGS / QPACK bytes.
      */
     suspend fun awaitIncomingPeerStream(): QuicStream? {
         while (true) {

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

@@ -0,0 +1,202 @@
+/*
+ * 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.stream.StreamId
+import com.vitorpamplona.quic.tls.InProcessTlsServer
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotEquals
+
+/**
+ * Regression coverage for the multiplexing-interop tear-down (audit-4 #3
+ * "slow consumer" overflow on peer-initiated uni streams).
+ *
+ * Reproduction of the production symptom:
+ *
+ *   The interop runner's `multiplexing` testcase opens many parallel
+ *   client-bidi GET streams via an H3 client. The H3 client opens its
+ *   three local uni streams (control + QPACK encoder + QPACK decoder)
+ *   per RFC 9114 §6.2.1 but **does not consume the server's three
+ *   counterpart peer-initiated uni streams**. Their bytes
+ *   (SETTINGS, dynamic-table inserts, ack signals) are routed by
+ *   [QuicConnectionParser] into each stream's bounded
+ *   `incomingChannel` (capacity 64). Once the QPACK encoder
+ *   stream's burst of dynamic-table inserts saturates it, the next
+ *   delivery trips the audit-4 #3 escape hatch:
+ *
+ *     conn.markClosedExternally("INTERNAL_ERROR: stream … consumer
+ *                               overflowed incoming channel
+ *                               (slow consumer)")
+ *
+ *   and the entire connection dies after ~4.5 s with zero requests
+ *   completed.
+ *
+ * The test pair below pins both halves of the contract:
+ *
+ *   - [pre_fix_no_consumer_overflows_and_tears_down_connection] — without
+ *     a consumer the connection MUST close with the documented reason.
+ *   - [drainPeerInitiatedUniStreamsIntoBlackHole_keeps_connection_alive] —
+ *     with [drainPeerInitiatedUniStreamsIntoBlackHole] running the
+ *     connection MUST stay CONNECTED and absorb the same byte volume.
+ *
+ * The fix philosophy (variant B from the prompt's three-way menu): the
+ * `:quic` library stays strict about backpressure — silently dropping
+ * app-data bytes is worse than failing fast. The new public helper is
+ * the explicit "I do not care about these particular peer streams"
+ * opt-in for an H3 GET client that runs with the QPACK dynamic table
+ * off. Default behaviour is unchanged.
+ */
+class PeerUniStreamDrainTest {
+    @Test
+    fun pre_fix_no_consumer_overflows_and_tears_down_connection() {
+        runBlocking {
+            val client = buildClient()
+            val pipe = buildPipe(client)
+            client.start()
+            pipe.drive(maxRounds = 16)
+            assertEquals(QuicConnection.Status.CONNECTED, client.status)
+
+            // Push 65 chunks on a server-initiated uni stream — one more
+            // than the per-stream incomingChannel capacity (64). The
+            // first 64 land; the 65th overflows trySend, sets
+            // QuicStream.overflowed, and the parser maps that to
+            // markClosedExternally with the documented reason.
+            //
+            // Each chunk fits comfortably under the per-stream receive
+            // limit (initialMaxStreamDataUni = 1 MiB) so the prior
+            // receive-limit guard does NOT fire — this test is about
+            // the *channel* overflow specifically, not flow control.
+            val streamId = StreamId.build(StreamId.Kind.SERVER_UNI, 0)
+            var offset = 0L
+            for (i in 0 until 65) {
+                val chunk = ByteArray(8) { (i + it).toByte() }
+                val frame = StreamFrame(streamId = streamId, offset = offset, data = chunk, fin = false)
+                offset += chunk.size.toLong()
+                val packet = pipe.buildServerApplicationDatagram(listOf(frame))
+                assertNotEquals(null, packet, "server has app keys after handshake")
+                feedDatagram(client, packet!!, nowMillis = 0L)
+            }
+
+            // The 65th chunk must have torn the connection down.
+            assertEquals(
+                QuicConnection.Status.CLOSED,
+                client.status,
+                "without a peer-uni-stream consumer the audit-4 #3 escape " +
+                    "hatch must fire on the 65th chunk",
+            )
+        }
+    }
+
+    @Test
+    fun drainPeerInitiatedUniStreamsIntoBlackHole_keeps_connection_alive() {
+        runBlocking {
+            val client = buildClient()
+            val pipe = buildPipe(client)
+            client.start()
+            pipe.drive(maxRounds = 16)
+            assertEquals(QuicConnection.Status.CONNECTED, client.status)
+
+            // Wire the explicit drainer BEFORE pushing the bytes.
+            val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+            try {
+                client.drainPeerInitiatedUniStreamsIntoBlackHole(scope)
+
+                // Same volume as the pre-fix test, except this time we go
+                // a long way past the channel capacity to prove sustained
+                // operation (4× the bound). If the drainer were absent the
+                // connection would have died at chunk 65; with the
+                // drainer reading them as fast as the parser delivers, no
+                // backpressure builds up.
+                val streamId = StreamId.build(StreamId.Kind.SERVER_UNI, 0)
+                var offset = 0L
+                for (i in 0 until 256) {
+                    val chunk = ByteArray(8) { (i + it).toByte() }
+                    val frame = StreamFrame(streamId = streamId, offset = offset, data = chunk, fin = false)
+                    offset += chunk.size.toLong()
+                    val packet = pipe.buildServerApplicationDatagram(listOf(frame))!!
+                    feedDatagram(client, packet, nowMillis = 0L)
+                    // Yield occasionally so the drainer coroutine actually
+                    // gets a chance to consume — feedDatagram is synchronous
+                    // and the drainer launched with Dispatchers.Default needs
+                    // a scheduling tick.
+                    if (i % 16 == 15) {
+                        withTimeout(2_000) { delay(1) }
+                    }
+                }
+                // Ensure the drainer has caught up before we sample status.
+                withTimeout(2_000) { delay(50) }
+
+                assertEquals(
+                    QuicConnection.Status.CONNECTED,
+                    client.status,
+                    "with drainPeerInitiatedUniStreamsIntoBlackHole the " +
+                        "connection must absorb arbitrary peer-uni traffic " +
+                        "without overflowing the channel; saw closeReason=" +
+                        "${client.closeReason}",
+                )
+            } finally {
+                scope.cancel()
+            }
+        }
+    }
+
+    private fun buildClient(): QuicConnection =
+        QuicConnection(
+            serverName = "example.test",
+            config = QuicConnectionConfig(),
+            tlsCertificateValidator =
+                com.vitorpamplona.quic.tls
+                    .PermissiveCertificateValidator(),
+        )
+
+    private fun buildPipe(client: QuicConnection): InMemoryQuicPipe {
+        val serverScid = ConnectionId.random(8)
+        val tlsServer =
+            InProcessTlsServer(
+                transportParameters =
+                    TransportParameters(
+                        initialMaxData = 10_000_000,
+                        initialMaxStreamDataBidiLocal = 1_000_000,
+                        initialMaxStreamDataBidiRemote = 1_000_000,
+                        initialMaxStreamDataUni = 1_000_000,
+                        initialMaxStreamsBidi = 16,
+                        initialMaxStreamsUni = 16,
+                        initialSourceConnectionId = serverScid.bytes,
+                        originalDestinationConnectionId = client.destinationConnectionId.bytes,
+                    ).encode(),
+            )
+        return InMemoryQuicPipe(
+            client = client,
+            initialDcid = client.destinationConnectionId.bytes,
+            serverScid = serverScid,
+            tlsServer = tlsServer,
+        )
+    }
+}