From 8f5280c9c3372cc5b4f82c3cd7c478418db615cf Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 25 Apr 2026 22:49:32 +0000 Subject: [PATCH] fix(quic): WebTransport peer-stream demux + flow-control + perf cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-2 audit found that without prefix-stripping on peer-initiated streams, the server's HTTP/3 control stream would deliver its SETTINGS frame to the application, breaking MoQ framing. Plus several flow-control + perf items the audit flagged. WtPeerStreamDemux: Spawned per WT session in QuicWebTransportSessionState.init. Routes peer streams by their leading varint(s): - HTTP/3 CONTROL stream (0x00) → drains internally, captures peer SETTINGS frame for inspection (peerSettings property). - QPACK encoder/decoder streams (0x02/0x03) → drained to /dev/null (we run with QPACK_MAX_TABLE_CAPACITY=0). - WebTransport unidirectional (0x54) + matching quarter session id → surfaces to app as a StrippedWtStream with the prefix stripped. - WebTransport bidi signal (0x41) + matching quarter session id → same. - Anything else → dropped per RFC 9114 §9. pollIncomingPeerStream is now @Deprecated; the recommended path is the incomingStrippedStreams flow on the session state. The nestsClient adapter now consumes the stripped flow and emits StrippedWtReadStreamAdapter through the existing WebTransportSession.incomingUniStreams API. WT_CLOSE_SESSION decoder + CapsuleReader: Added stateful capsule reader that yields parsed WtCloseSession (errorCode + reason) from the CONNECT bidi. Wires into the future close notification path; not yet bound to status flip but the parser is in place. Flow-control fixes from the audit: - Per-direction receive window in getOrCreatePeerStreamLocked. SERVER_UNI streams now use initialMaxStreamDataUni instead of bidi-remote. - appendFlowControlUpdates picks per-direction window matching the stream's StreamId.kindOf. - MAX_DATA tracking via QuicConnection.advertisedMaxData high-water mark. Previously the writer emitted a fresh MAX_DATA on every outbound packet once the threshold was crossed (spam). Now: only when the new value strictly exceeds advertisedMaxData. Stream iteration round-robin: Writer was iterating streamsLocked() in insertion order, starving streams created later under MTU pressure. Added streamRoundRobinStart counter on QuicConnection that advances after each drain. SendBuffer alias defense: takeChunk's fast path used to return the head ByteArray reference directly when consuming the whole chunk — caller-owned byte arrays could be mutated in-place. Now always copies, since downstream encoders pass the bytes to AEAD.seal which assumes immutability. UdpSocket cleanup: - readBuf shrunk from 65 KiB to 2 KiB. QUIC datagrams cap at MTU; the old 65 KiB allocation was per-connection waste with no benefit. - Removed pointless synchronized(readBuf) — only the read loop touches it. All :quic:jvmTest + :nestsClient:jvmTest pass. https://claude.ai/code/session_01EC1tfXfap8k8GyKvrxkxZx --- .../transport/QuicWebTransportFactory.kt | 19 +- .../quic/connection/QuicConnection.kt | 25 +- .../quic/connection/QuicConnectionWriter.kt | 57 ++-- .../vitorpamplona/quic/stream/SendBuffer.kt | 13 +- .../QuicWebTransportSessionState.kt | 44 ++++ .../quic/webtransport/WtCapsule.kt | 71 +++++ .../quic/webtransport/WtPeerStreamDemux.kt | 244 ++++++++++++++++++ .../vitorpamplona/quic/transport/UdpSocket.kt | 24 +- 8 files changed, 462 insertions(+), 35 deletions(-) create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt diff --git a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt index d32e56d80..783ac3117 100644 --- a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt +++ b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt @@ -219,10 +219,14 @@ class QuicWebTransportSession( override fun incomingUniStreams(): Flow = flow { - while (state.isOpen) { - val s = state.pollIncomingPeerStream() - if (s != null) emit(QuicReadStreamAdapter(s)) - kotlinx.coroutines.delay(10) + // Surface only unidirectional WT streams whose prefix bytes + // (0x54 + quarter session id) have been stripped. The H3 control + // stream and QPACK encoder/decoder streams are drained internally + // by the demux and never reach here. + state.incomingStrippedStreams.collect { stripped -> + if (stripped.isUnidirectional) { + emit(StrippedWtReadStreamAdapter(stripped)) + } } } @@ -270,3 +274,10 @@ private class QuicReadStreamAdapter( ) : WebTransportReadStream { override fun incoming(): Flow = stream.incoming } + +/** Adapter for a WT peer-initiated uni stream whose prefix has been stripped. */ +private class StrippedWtReadStreamAdapter( + private val stripped: com.vitorpamplona.quic.webtransport.StrippedWtStream, +) : WebTransportReadStream { + override fun incoming(): Flow = stripped.data +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index 3e439d027..ca3fa4a84 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -102,6 +102,21 @@ class QuicConnection( private val streams = mutableMapOf() private var nextLocalBidiIndex: Long = 0L private var nextLocalUniIndex: Long = 0L + + /** + * The connection-level receive limit we've currently advertised to the + * peer. Tracks the high-water mark of the most recent MAX_DATA frame we + * sent. The writer only emits a new MAX_DATA when the new value exceeds + * this — prevents spamming the peer with redundant updates. + */ + internal var advertisedMaxData: Long = config.initialMaxData + + /** + * Round-robin starting index for the writer's stream-drain iteration. + * Without rotation, streams created earlier always drain first under MTU + * pressure, starving later streams indefinitely. + */ + internal var streamRoundRobinStart: Int = 0 private val pendingDatagrams = ArrayDeque() private val incomingDatagrams = ArrayDeque() private var sendConnectionFlowCredit: Long = 0L @@ -293,7 +308,15 @@ class QuicConnection( } val stream = QuicStream(id, direction) stream.sendCredit = 0L - stream.receiveLimit = config.initialMaxStreamDataBidiRemote + // Pick the local receive-limit appropriate for the stream's direction + // — peer-bidi → we advertised initialMaxStreamDataBidiRemote; + // peer-uni → we advertised initialMaxStreamDataUni. + stream.receiveLimit = + when (StreamId.kindOf(id)) { + StreamId.Kind.SERVER_UNI, StreamId.Kind.CLIENT_UNI -> config.initialMaxStreamDataUni + StreamId.Kind.SERVER_BIDI -> config.initialMaxStreamDataBidiRemote + StreamId.Kind.CLIENT_BIDI -> config.initialMaxStreamDataBidiLocal + } streams[id] = stream newPeerStreams.addLast(stream) return stream diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index caf282605..af758d2e9 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -257,20 +257,26 @@ private fun buildApplicationPacket( if (frames.size >= 16) break } - // Drain stream send buffers — round-robin keeping packet under MTU and - // honoring per-stream + connection-level send credit (RFC 9000 §4). + // Drain stream send buffers — round-robin starting from a rotating index + // so streams created earlier don't starve streams created later under MTU + // pressure. Honors per-stream send credit (RFC 9000 §4). var packetBudget = 1100 - for ((id, stream) in conn.streamsLocked()) { - if (packetBudget <= 64) break - // Per-stream credit: how many more bytes is the peer willing to receive on this stream? - val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L) - if (streamRemaining <= 0L && !stream.send.finPending) continue - val maxBytes = minOf(packetBudget - 32, streamRemaining.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) - val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue - if (chunk.data.isNotEmpty() || chunk.fin) { - frames += StreamFrame(streamId = id, offset = chunk.offset, data = chunk.data, fin = chunk.fin, explicitLength = true) - packetBudget -= chunk.data.size + 32 + val streamsList = conn.streamsLocked().entries.toList() + if (streamsList.isNotEmpty()) { + val start = conn.streamRoundRobinStart % streamsList.size + for (i in streamsList.indices) { + if (packetBudget <= 64) break + val (id, stream) = streamsList[(start + i) % streamsList.size] + val streamRemaining = (stream.sendCredit - stream.send.sentOffset).coerceAtLeast(0L) + if (streamRemaining <= 0L && !stream.send.finPending) continue + val maxBytes = minOf(packetBudget - 32, streamRemaining.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) + val chunk = stream.send.takeChunk(maxBytes = maxBytes) ?: continue + if (chunk.data.isNotEmpty() || chunk.fin) { + frames += StreamFrame(streamId = id, offset = chunk.offset, data = chunk.data, fin = chunk.fin, explicitLength = true) + packetBudget -= chunk.data.size + 32 + } } + conn.streamRoundRobinStart = (start + 1) % streamsList.size } if (frames.isEmpty()) return null @@ -302,8 +308,20 @@ private fun appendFlowControlUpdates( for ((id, stream) in conn.streamsLocked()) { val rcv = stream.receive.contiguousEnd() if (rcv == 0L) continue + // Pick the per-direction window matching the stream kind so a + // uni-only deployment with bidi=0 doesn't accidentally use the bidi + // window and vice versa. + val window = + when ( + com.vitorpamplona.quic.stream.StreamId + .kindOf(id) + ) { + com.vitorpamplona.quic.stream.StreamId.Kind.SERVER_UNI -> cfg.initialMaxStreamDataUni + com.vitorpamplona.quic.stream.StreamId.Kind.SERVER_BIDI -> cfg.initialMaxStreamDataBidiRemote + com.vitorpamplona.quic.stream.StreamId.Kind.CLIENT_BIDI -> cfg.initialMaxStreamDataBidiLocal + com.vitorpamplona.quic.stream.StreamId.Kind.CLIENT_UNI -> cfg.initialMaxStreamDataUni + } // Re-credit when consumed > half the advertised window. - val window = cfg.initialMaxStreamDataBidiRemote.coerceAtLeast(cfg.initialMaxStreamDataUni) if (rcv >= stream.receiveLimit - window / 2) { val newLimit = rcv + window if (newLimit > stream.receiveLimit) { @@ -313,9 +331,14 @@ private fun appendFlowControlUpdates( } totalRecvAdvanced += rcv } - // Connection-level credit: when sum of contiguousEnd across all streams - // exceeds half the connection-level limit, re-grant. - if (totalRecvAdvanced > 0L && totalRecvAdvanced >= cfg.initialMaxData / 2) { - frames += MaxDataFrame(totalRecvAdvanced + cfg.initialMaxData) + // Connection-level: only re-grant when the new total would exceed our + // currently-advertised limit. Without this we'd emit a fresh MaxDataFrame + // on every outbound packet once the threshold was crossed. + if (totalRecvAdvanced > 0L && totalRecvAdvanced + cfg.initialMaxData / 2 >= conn.advertisedMaxData) { + val newLimit = totalRecvAdvanced + cfg.initialMaxData + if (newLimit > conn.advertisedMaxData) { + conn.advertisedMaxData = newLimit + frames += MaxDataFrame(newLimit) + } } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt index f78d52c39..07d983cd3 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt @@ -83,8 +83,17 @@ class SendBuffer { val head = chunks.first() val available = head.size - headOffset if (available <= cap) { - // Hand out the rest of the head chunk. - data = if (headOffset == 0) head else head.copyOfRange(headOffset, head.size) + // Hand out the rest of the head chunk. Always copy: the caller's + // ByteArray (passed to enqueue) MUST stay opaque to the rest of + // the stack, since downstream encoders eventually pass it to + // AEAD.seal which assumes immutability for the duration of the + // encryption call. + data = + if (headOffset == 0 && head.size == available) { + head.copyOf() + } else { + head.copyOfRange(headOffset, head.size) + } chunks.removeFirst() headOffset = 0 pendingBytes -= available diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt index f85123791..501e3c7ef 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/QuicWebTransportSessionState.kt @@ -23,6 +23,11 @@ package com.vitorpamplona.quic.webtransport import com.vitorpamplona.quic.connection.QuicConnection import com.vitorpamplona.quic.connection.QuicConnectionDriver import com.vitorpamplona.quic.stream.QuicStream +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.launch /** * Container that bundles a [QuicConnection], its [QuicConnectionDriver], and @@ -36,11 +41,42 @@ class QuicWebTransportSessionState( val connection: QuicConnection, val driver: QuicConnectionDriver, val connectStreamId: Long, + /** + * Scope used to spawn the peer-stream demux. Defaults to a SupervisorJob + * so a single misbehaving stream doesn't tear down the session. + */ + private val scope: CoroutineScope = + CoroutineScope(SupervisorJob() + Dispatchers.IO), ) { /** True until [close] is called or the underlying QUIC connection terminates. */ val isOpen: Boolean get() = connection.status == QuicConnection.Status.CONNECTED + private val demux: WtPeerStreamDemux = WtPeerStreamDemux(connectStreamId, scope) + + init { + // Spawn the dispatcher that routes peer-initiated streams through the + // demux. Without this, the server's CONTROL stream (carrying SETTINGS) + // would be handed to the application, which would interpret SETTINGS + // bytes as MoQ frames and break framing. + scope.launch { + while (connection.status != QuicConnection.Status.CLOSED) { + val s = connection.pollIncomingPeerStream() + if (s != null) { + demux.process(s) + } else { + kotlinx.coroutines.delay(5) + } + } + } + } + + /** Server SETTINGS once received on the H3 control stream; null until then. */ + val peerSettings get() = demux.peerSettings + + /** Flow of peer-initiated WT streams whose framing prefix has been stripped. */ + val incomingStrippedStreams: Flow get() = demux.incomingStrippedStreams + /** Open a new client-initiated bidirectional WebTransport stream. */ suspend fun openBidiStream(): QuicStream { val s = connection.openBidiStream() @@ -72,6 +108,14 @@ class QuicWebTransportSessionState( return decoded.payload } + /** + * @deprecated Use [incomingStrippedStreams] which yields streams whose + * framing prefix (CONTROL/QPACK/WT type bytes + quarter session id) has + * already been stripped. Direct callers of this would receive raw peer + * streams including the server's CONTROL stream, with SETTINGS bytes + * interpreted as application data. + */ + @Deprecated("Use incomingStrippedStreams instead", ReplaceWith("incomingStrippedStreams")) suspend fun pollIncomingPeerStream(): QuicStream? = connection.pollIncomingPeerStream() suspend fun close( diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt index fe5fe04f9..e956961d6 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtCapsule.kt @@ -53,3 +53,74 @@ fun encodeCloseSessionCapsule( body.writeBytes(reason.encodeToByteArray()) return encodeCapsule(WtCapsuleType.WT_CLOSE_SESSION, body.toByteArray()) } + +/** Parsed WT_CLOSE_SESSION capsule. */ +data class WtCloseSession( + val errorCode: Int, + val reason: String, +) + +/** + * Stateful capsule reader. Capsules on the WT CONNECT bidi stream are + * `(varint type)(varint length)(body)`. Feed bytes via [push]; drain + * complete capsules via [next]. + */ +class CapsuleReader { + private var buf: ByteArray = ByteArray(0) + private var pos: Int = 0 + + fun push(bytes: ByteArray) { + if (bytes.isEmpty()) return + if (pos > 0) { + buf = buf.copyOfRange(pos, buf.size) + pos = 0 + } + val combined = ByteArray(buf.size + bytes.size) + buf.copyInto(combined, 0) + bytes.copyInto(combined, buf.size) + buf = combined + } + + /** + * Pop the next complete capsule, or null if the buffer doesn't contain + * one yet. Returns either a [WtCloseSession] or a raw `(type, body)` + * pair for unknown capsule types. + */ + fun next(): Any? { + val typeRes = + com.vitorpamplona.quic.Varint + .decode(buf, pos) ?: return null + val typeEnd = pos + typeRes.bytesConsumed + val lenRes = + com.vitorpamplona.quic.Varint + .decode(buf, typeEnd) ?: return null + val bodyStart = typeEnd + lenRes.bytesConsumed + val len = lenRes.value + if (len < 0 || len > Int.MAX_VALUE.toLong()) { + throw com.vitorpamplona.quic.QuicCodecException("capsule length out of range: $len") + } + val bodyEnd = bodyStart + len.toInt() + if (bodyEnd > buf.size) return null + val body = buf.copyOfRange(bodyStart, bodyEnd) + pos = bodyEnd + return when (typeRes.value) { + WtCapsuleType.WT_CLOSE_SESSION -> { + if (body.size < 4) { + WtCloseSession(0, "") + } else { + val errorCode = + ((body[0].toInt() and 0xFF) shl 24) or + ((body[1].toInt() and 0xFF) shl 16) or + ((body[2].toInt() and 0xFF) shl 8) or + (body[3].toInt() and 0xFF) + val reason = body.copyOfRange(4, body.size).decodeToString() + WtCloseSession(errorCode, reason) + } + } + + else -> { + typeRes.value to body + } + } + } +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt new file mode 100644 index 000000000..5d9e96ed9 --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/webtransport/WtPeerStreamDemux.kt @@ -0,0 +1,244 @@ +/* + * 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.webtransport + +import com.vitorpamplona.quic.Varint +import com.vitorpamplona.quic.http3.Http3Frame +import com.vitorpamplona.quic.http3.Http3FrameReader +import com.vitorpamplona.quic.http3.Http3Settings +import com.vitorpamplona.quic.http3.Http3StreamType +import com.vitorpamplona.quic.stream.QuicStream +import com.vitorpamplona.quic.stream.StreamId +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.consumeAsFlow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.launch + +/** + * A peer-initiated WebTransport stream whose framing prefix has already been + * stripped. The [data] flow yields only application-level bytes. + */ +class StrippedWtStream( + val streamId: Long, + val isUnidirectional: Boolean, + val data: Flow, +) + +/** + * Demultiplexes peer-initiated streams by their leading varint(s): + * + * - HTTP/3 CONTROL stream (0x00) — drained internally, SETTINGS captured. + * - QPACK encoder/decoder streams (0x02 / 0x03) — drained (we run with + * dynamic table off). + * - WT unidirectional (0x54) + matching quarter session id — surfaced. + * - WT bidirectional signal (0x41) + matching quarter session id — surfaced. + * - Anything else — dropped per RFC 9114 §9. + * + * Without this, the server's CONTROL stream would deliver its SETTINGS frame + * bytes directly to MoQ as application data — a live-interop break. + */ +class WtPeerStreamDemux( + private val expectedConnectStreamId: Long, + private val scope: CoroutineScope, +) { + private val readyStreams = Channel(Channel.UNLIMITED) + + @Volatile + var peerSettings: Http3Settings? = null + private set + + val incomingStrippedStreams: Flow = readyStreams.consumeAsFlow() + + /** + * Begin processing a peer-initiated stream. Caller must invoke this for + * every stream returned by [com.vitorpamplona.quic.connection.QuicConnection.pollIncomingPeerStream]. + */ + fun process(stream: QuicStream) { + scope.launch { route(stream) } + } + + 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() + val flowIterator = stream.incoming + val chunkChannel = Channel(Channel.UNLIMITED) + scope.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 { + if (StreamId.isUnidirectional(stream.streamId)) { + val streamType = readVarintFromPending() ?: return + when (streamType) { + Http3StreamType.CONTROL -> { + drainControlStream(pending, chunkChannel) + } + + 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 + } + 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() ?: 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 (_: Throwable) { + // peer closed mid-prefix or framing error — drop quietly + } + } + + private suspend fun drainControlStream( + pending: ArrayDeque, + chunkChannel: Channel, + ) { + val reader = Http3FrameReader() + // Push whatever we already buffered. + while (pending.isNotEmpty()) reader.push(pending.removeFirst()) + consumeFrames(reader) + for (chunk in chunkChannel) { + reader.push(chunk) + consumeFrames(reader) + } + } + + private fun consumeFrames(reader: Http3FrameReader) { + while (true) { + val frame = reader.next() ?: return + when (frame) { + is Http3Frame.Settings -> peerSettings = frame.settings + + is Http3Frame.Goaway -> Unit + + // no new requests; we don't enforce yet + else -> Unit + } + } + } + + private suspend fun drainBlackHole(chunkChannel: Channel) { + @Suppress("UNUSED_VARIABLE") + for (discarded in chunkChannel) { + // intentionally discarded — stream type is one we don't process + } + } + + private fun emitStripped( + stream: QuicStream, + pending: ArrayDeque, + chunkChannel: Channel, + isUni: Boolean, + ) { + val prebuffered = pending.toList() + pending.clear() + val data: Flow = + flow { + for (b in prebuffered) if (b.isNotEmpty()) emit(b) + for (chunk in chunkChannel) if (chunk.isNotEmpty()) emit(chunk) + } + readyStreams.trySend( + StrippedWtStream( + streamId = stream.streamId, + isUnidirectional = isUni, + data = data, + ), + ) + } +} + +private fun flatten(chunks: ArrayDeque): ByteArray { + var total = 0 + for (c in chunks) total += c.size + val out = ByteArray(total) + var pos = 0 + for (c in chunks) { + c.copyInto(out, pos) + pos += c.size + } + return out +} + +private fun consumeFromPending( + pending: ArrayDeque, + bytes: Int, +) { + var remaining = bytes + while (remaining > 0 && pending.isNotEmpty()) { + val head = pending.first() + if (head.size <= remaining) { + pending.removeFirst() + remaining -= head.size + } else { + pending[0] = head.copyOfRange(remaining, head.size) + remaining = 0 + } + } +} diff --git a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt index 3c7c178ad..fdd6420d5 100644 --- a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt +++ b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt @@ -43,7 +43,11 @@ actual class UdpSocket private constructor( private val remote: InetSocketAddress, ) { private val closed = AtomicBoolean(false) - private val readBuf = ByteBuffer.allocate(MAX_DATAGRAM_SIZE) + + // Sized to typical Ethernet MTU + a bit; QUIC tops out at ~1500 in practice + // and any larger inbound frame is dropped as malformed anyway. The previous + // 64 KiB buffer was wasteful per connection. + private val readBuf = ByteBuffer.allocate(2048) actual val localPort: Int get() = (channel.localAddress as InetSocketAddress).port @@ -59,14 +63,14 @@ actual class UdpSocket private constructor( withContext(Dispatchers.IO) { if (closed.get()) return@withContext null try { - synchronized(readBuf) { - readBuf.clear() - channel.receive(readBuf) ?: return@withContext null - readBuf.flip() - val out = ByteArray(readBuf.remaining()) - readBuf.get(out) - out - } + // No synchronized — only the read loop touches readBuf, by + // contract. The previous synchronized block was pointless. + readBuf.clear() + channel.receive(readBuf) ?: return@withContext null + readBuf.flip() + val out = ByteArray(readBuf.remaining()) + readBuf.get(out) + out } catch (_: ClosedChannelException) { null } @@ -83,8 +87,6 @@ actual class UdpSocket private constructor( } actual companion object { - private const val MAX_DATAGRAM_SIZE: Int = 65535 - actual suspend fun connect( host: String, port: Int,