From 52ab84acfe00bd52da18c71bcb13c1b45884abcf Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 25 Apr 2026 20:50:26 +0000 Subject: [PATCH] =?UTF-8?q?feat(quic):=20Phase=20D-H=20=E2=80=94=20QuicCon?= =?UTF-8?q?nection=20orchestrator?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire together TLS, packet codec, and stream buffers into a single client QuicConnection that drives the handshake and 1-RTT exchange. - QuicConnection owns per-encryption-level state (LevelState): packet number space, ACK tracker, CRYPTO send + receive buffers, send + receive packet protection. Initial keys are installed at construction from a random DCID. - TlsSecretsListener inside the connection installs handshake + application keys as the TLS state machine derives them. - AckTracker maintains a sorted disjoint-range list of received packet numbers and emits RFC 9000 §19.3-shaped ACK frames newest-first. - SendBuffer + per-stream QuicStream allow application code to enqueue bytes and FIN; the connection's writer drains them into STREAM frames. - QuicConnectionParser dispatches inbound CRYPTO/STREAM/DATAGRAM/MAX_*/ACK/ CONNECTION_CLOSE frames into the right state. Coalesced packets in a single datagram are demultiplexed. - QuicConnectionWriter builds outbound datagrams that coalesce Initial + Handshake + 1-RTT packets, pads Initial datagrams to 1200 bytes, and drains stream send queues round-robin under a soft MTU budget. - TransportParameters codec covers all RFC 9000 §18.2 + RFC 9221 fields and is exchanged via the TLS quic_transport_parameters extension. - PacketProtectionBuilder maps a TLS traffic secret + cipher suite to the AEAD + HP triple via RFC 9001's `quic key`/`quic iv`/`quic hp` labels. https://claude.ai/code/session_01EC1tfXfap8k8GyKvrxkxZx --- .../quic/connection/EncryptionLevel.kt | 40 +++ .../quic/connection/LevelState.kt | 34 +++ .../connection/PacketProtectionBuilder.kt | 73 +++++ .../quic/connection/QuicConnection.kt | 268 ++++++++++++++++++ .../quic/connection/QuicConnectionConfig.kt | 46 +++ .../quic/connection/QuicConnectionParser.kt | 230 +++++++++++++++ .../quic/connection/QuicConnectionWriter.kt | 246 ++++++++++++++++ .../vitorpamplona/quic/recovery/AckTracker.kt | 115 ++++++++ .../vitorpamplona/quic/stream/QuicStream.kt | 67 +++++ .../vitorpamplona/quic/stream/SendBuffer.kt | 83 ++++++ 10 files changed, 1202 insertions(+) create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/EncryptionLevel.kt create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/PacketProtectionBuilder.kt create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt create mode 100644 quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/EncryptionLevel.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/EncryptionLevel.kt new file mode 100644 index 000000000..b61430a3e --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/EncryptionLevel.kt @@ -0,0 +1,40 @@ +/* + * 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.Aead + +/** Per-direction packet protection material at one encryption level. */ +class PacketProtection( + val aead: Aead, + val key: ByteArray, + val iv: ByteArray, + val hp: com.vitorpamplona.quic.crypto.HeaderProtection, + val hpKey: ByteArray, +) + +/** All four encryption levels we ever see in a QUIC client connection. */ +enum class EncryptionLevel(val space: PacketNumberSpace) { + INITIAL(PacketNumberSpace.INITIAL), + HANDSHAKE(PacketNumberSpace.HANDSHAKE), + APPLICATION(PacketNumberSpace.APPLICATION), + ; +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt new file mode 100644 index 000000000..46c219f51 --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/LevelState.kt @@ -0,0 +1,34 @@ +/* + * 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.ReceiveBuffer +import com.vitorpamplona.quic.stream.SendBuffer + +/** Per-encryption-level state owned by [QuicConnection]. */ +class LevelState { + val pnSpace = PacketNumberSpaceState() + val ackTracker = com.vitorpamplona.quic.recovery.AckTracker() + val cryptoSend = SendBuffer() + val cryptoReceive = ReceiveBuffer() + var sendProtection: PacketProtection? = null + var receiveProtection: PacketProtection? = null +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/PacketProtectionBuilder.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/PacketProtectionBuilder.kt new file mode 100644 index 000000000..cda223847 --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/PacketProtectionBuilder.kt @@ -0,0 +1,73 @@ +/* + * 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.Aead +import com.vitorpamplona.quic.crypto.Aes128Gcm +import com.vitorpamplona.quic.crypto.AesEcbHeaderProtection +import com.vitorpamplona.quic.crypto.HeaderProtection +import com.vitorpamplona.quic.crypto.PlatformAesOneBlock +import com.vitorpamplona.quic.tls.TlsConstants +import com.vitorpamplona.quic.tls.deriveQuicKeys + +/** + * Build [PacketProtection] for one direction at one encryption level given a + * TLS traffic secret + TLS cipher-suite identifier. The QUIC labels in + * RFC 9001 §5.1 (`quic key`, `quic iv`, `quic hp`) drive the expansion. + * + * For Phase B–K we only support TLS_AES_128_GCM_SHA256 (16/12/16) and + * TLS_CHACHA20_POLY1305_SHA256 — the SHA-256 suites; nests speaks both. + */ +fun packetProtectionFromSecret( + cipherSuite: Int, + secret: ByteArray, +): PacketProtection { + val (aead, keyLen, ivLen, hpLen, hp) = + when (cipherSuite) { + TlsConstants.CIPHER_TLS_AES_128_GCM_SHA256 -> + ProtectionParams( + aead = Aes128Gcm, + keyLen = 16, + ivLen = 12, + hpLen = 16, + hp = AesEcbHeaderProtection(PlatformAesOneBlock), + ) + TlsConstants.CIPHER_TLS_CHACHA20_POLY1305_SHA256 -> + ProtectionParams( + aead = com.vitorpamplona.quic.crypto.ChaCha20Poly1305Aead, + keyLen = 32, + ivLen = 12, + hpLen = 32, + hp = com.vitorpamplona.quic.crypto.ChaCha20HeaderProtection(com.vitorpamplona.quic.crypto.PlatformChaCha20Block), + ) + else -> error("unsupported cipher suite 0x${cipherSuite.toString(16)}") + } + val keys = deriveQuicKeys(secret, keyLen, ivLen, hpLen) + return PacketProtection(aead, keys.key, keys.iv, hp, keys.hp) +} + +private data class ProtectionParams( + val aead: Aead, + val keyLen: Int, + val ivLen: Int, + val hpLen: Int, + val hp: HeaderProtection, +) diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt new file mode 100644 index 000000000..e9d374eff --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -0,0 +1,268 @@ +/* + * 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.QuicWriter +import com.vitorpamplona.quic.crypto.Aes128Gcm +import com.vitorpamplona.quic.crypto.AesEcbHeaderProtection +import com.vitorpamplona.quic.crypto.InitialSecrets +import com.vitorpamplona.quic.crypto.PlatformAesOneBlock +import com.vitorpamplona.quic.frame.ConnectionCloseFrame +import com.vitorpamplona.quic.frame.CryptoFrame +import com.vitorpamplona.quic.frame.DatagramFrame +import com.vitorpamplona.quic.frame.Frame +import com.vitorpamplona.quic.frame.PingFrame +import com.vitorpamplona.quic.frame.StreamFrame +import com.vitorpamplona.quic.frame.decodeFrames +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 com.vitorpamplona.quic.packet.ShortHeaderPacket +import com.vitorpamplona.quic.packet.ShortHeaderPlaintextPacket +import com.vitorpamplona.quic.stream.QuicStream +import com.vitorpamplona.quic.stream.StreamId +import com.vitorpamplona.quic.tls.TlsClient +import com.vitorpamplona.quic.tls.TlsConstants +import com.vitorpamplona.quic.tls.TlsSecretsListener + +/** + * QUIC v1 client connection. The orchestrator that owns: + * + * - the TLS 1.3 client (which produces CRYPTO bytes per encryption level) + * - the per-level packet protection material + * - per-level packet number spaces + ACK trackers + * - per-stream send/receive buffers + * - inbound datagram queue + * + * The connection is driven by two callbacks: + * + * [feedDatagram] — feed an inbound UDP datagram. Decrypts every contained + * QUIC packet, runs CRYPTO bytes through the TLS state + * machine, dispatches STREAM/DATAGRAM/ACK/MAX_* frames to + * the right stream, and updates ACK state. + * + * [drainOutbound] — pull the next UDP datagram (or null if nothing to send). + * Coalesces ACK + CRYPTO + STREAM + DATAGRAM frames into + * a single packet (or two, if Initial + Handshake). + * + * The actual UDP socket I/O is in the higher-level [QuicConnectionDriver]. + */ +class QuicConnection( + val serverName: String, + val config: QuicConnectionConfig, + val tlsCertificateValidator: com.vitorpamplona.quic.tls.CertificateValidator? = null, + val nowMillis: () -> Long = { kotlin.time.Clock.System.now().toEpochMilliseconds() }, + val alpnList: List = listOf(TlsConstants.ALPN_H3), +) { + val sourceConnectionId: ConnectionId = ConnectionId.random(8) + var destinationConnectionId: ConnectionId = ConnectionId.random(8) + internal set + val originalDestinationConnectionId: ConnectionId = destinationConnectionId + + val initial = LevelState() + val handshake = LevelState() + val application = LevelState() + + var handshakeComplete: Boolean = false + private set + + var peerTransportParameters: TransportParameters? = null + private set + + enum class Status { HANDSHAKING, CONNECTED, CLOSING, CLOSED } + + var status: Status = Status.HANDSHAKING + internal set + + /** App-level error code for graceful close. */ + var closeReason: String? = null + private set + var closeErrorCode: Long = 0L + private set + + private val streams = mutableMapOf() + private var nextLocalBidiIndex: Long = 0L + private var nextLocalUniIndex: Long = 0L + private val pendingDatagrams = ArrayDeque() + private val incomingDatagrams = ArrayDeque() + private var sendConnectionFlowCredit: Long = 0L + private var receiveConnectionFlowLimit: Long = config.initialMaxData + + /** Streams the peer has opened that we haven't surfaced yet. */ + private val newPeerStreams = ArrayDeque() + + private val tlsListener = + object : TlsSecretsListener { + override fun onHandshakeKeysReady( + cipherSuite: Int, + clientSecret: ByteArray, + serverSecret: ByteArray, + ) { + handshake.sendProtection = packetProtectionFromSecret(cipherSuite, clientSecret) + handshake.receiveProtection = packetProtectionFromSecret(cipherSuite, serverSecret) + } + + override fun onApplicationKeysReady( + cipherSuite: Int, + clientSecret: ByteArray, + serverSecret: ByteArray, + ) { + application.sendProtection = packetProtectionFromSecret(cipherSuite, clientSecret) + application.receiveProtection = packetProtectionFromSecret(cipherSuite, serverSecret) + } + + override fun onHandshakeComplete() { + handshakeComplete = true + if (status == Status.HANDSHAKING) status = Status.CONNECTED + applyPeerTransportParameters() + } + } + + val tls: TlsClient = + TlsClient( + serverName = serverName, + transportParameters = buildLocalTransportParameters().encode(), + secretsListener = tlsListener, + certificateValidator = tlsCertificateValidator, + ) + + init { + // Install Initial keys based on the random destination CID we just generated. + val proto = InitialSecrets.derive(destinationConnectionId.bytes) + val hp = AesEcbHeaderProtection(PlatformAesOneBlock) + initial.sendProtection = PacketProtection(Aes128Gcm, proto.clientKey, proto.clientIv, hp, proto.clientHp) + initial.receiveProtection = PacketProtection(Aes128Gcm, proto.serverKey, proto.serverIv, hp, proto.serverHp) + } + + /** Begin the handshake — emits ClientHello into Initial CRYPTO. */ + fun start() { + tls.start() + // Drain ClientHello bytes into the Initial-level CRYPTO send buffer. + tls.pollOutbound(TlsClient.Level.INITIAL)?.let { initial.cryptoSend.enqueue(it) } + } + + private fun buildLocalTransportParameters(): TransportParameters = + TransportParameters( + initialMaxData = config.initialMaxData, + initialMaxStreamDataBidiLocal = config.initialMaxStreamDataBidiLocal, + initialMaxStreamDataBidiRemote = config.initialMaxStreamDataBidiRemote, + initialMaxStreamDataUni = config.initialMaxStreamDataUni, + initialMaxStreamsBidi = config.initialMaxStreamsBidi, + initialMaxStreamsUni = config.initialMaxStreamsUni, + maxIdleTimeoutMillis = config.maxIdleTimeoutMillis, + maxUdpPayloadSize = config.maxUdpPayloadSize, + ackDelayExponent = config.ackDelayExponent, + maxAckDelay = config.maxAckDelay, + activeConnectionIdLimit = config.activeConnectionIdLimit, + initialSourceConnectionId = sourceConnectionId.bytes, + maxDatagramFrameSize = config.maxDatagramFrameSize, + ) + + private fun applyPeerTransportParameters() { + val raw = tls.peerTransportParameters ?: return + val tp = TransportParameters.decode(raw) + peerTransportParameters = tp + sendConnectionFlowCredit = tp.initialMaxData ?: 0L + // Update each open stream's send credit per direction. + for ((id, stream) in streams) { + stream.sendCredit = + when (StreamId.kindOf(id)) { + StreamId.Kind.CLIENT_BIDI, StreamId.Kind.SERVER_BIDI -> + if (StreamId.isClientInitiated(id)) tp.initialMaxStreamDataBidiRemote ?: 0L else tp.initialMaxStreamDataBidiLocal ?: 0L + StreamId.Kind.CLIENT_UNI, StreamId.Kind.SERVER_UNI -> + tp.initialMaxStreamDataUni ?: 0L + } + } + } + + /** Allocate a new client-initiated bidirectional stream. */ + fun openBidiStream(): QuicStream { + val id = StreamId.build(StreamId.Kind.CLIENT_BIDI, nextLocalBidiIndex++) + val stream = QuicStream(id, QuicStream.Direction.BIDIRECTIONAL) + stream.sendCredit = peerTransportParameters?.initialMaxStreamDataBidiRemote ?: config.initialMaxStreamDataBidiRemote + stream.receiveLimit = config.initialMaxStreamDataBidiLocal + streams[id] = stream + return stream + } + + /** Allocate a new client-initiated unidirectional (write-only) stream. */ + fun openUniStream(): QuicStream { + val id = StreamId.build(StreamId.Kind.CLIENT_UNI, nextLocalUniIndex++) + val stream = QuicStream(id, QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE) + stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni + stream.receiveLimit = 0L // can't receive + streams[id] = stream + return stream + } + + fun pollIncomingPeerStream(): QuicStream? = newPeerStreams.removeFirstOrNull() + + fun streamById(id: Long): QuicStream? = streams[id] + + fun queueDatagram(payload: ByteArray) { + pendingDatagrams.addLast(payload) + } + + fun pollIncomingDatagram(): ByteArray? = incomingDatagrams.removeFirstOrNull() + + /** Initiate a graceful close. */ + fun close( + errorCode: Long, + reason: String, + ) { + if (status == Status.CLOSED || status == Status.CLOSING) return + closeErrorCode = errorCode + closeReason = reason + status = Status.CLOSING + } + + internal fun getOrCreatePeerStream(id: Long): QuicStream { + streams[id]?.let { return it } + val direction = + when (StreamId.kindOf(id)) { + StreamId.Kind.CLIENT_BIDI, StreamId.Kind.SERVER_BIDI -> QuicStream.Direction.BIDIRECTIONAL + StreamId.Kind.SERVER_UNI -> QuicStream.Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL + StreamId.Kind.CLIENT_UNI -> QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE + } + val stream = QuicStream(id, direction) + stream.sendCredit = 0L + stream.receiveLimit = config.initialMaxStreamDataBidiRemote + streams[id] = stream + newPeerStreams.addLast(stream) + return stream + } + + /** Returns the level state for [level]. */ + fun levelState(level: EncryptionLevel): LevelState = + when (level) { + EncryptionLevel.INITIAL -> initial + EncryptionLevel.HANDSHAKE -> handshake + EncryptionLevel.APPLICATION -> application + } + + fun streamsView(): Map = streams + + fun pendingDatagramsView(): ArrayDeque = pendingDatagrams + + fun incomingDatagramsBuffer(): ArrayDeque = incomingDatagrams +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt new file mode 100644 index 000000000..0e33ee2fc --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt @@ -0,0 +1,46 @@ +/* + * 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 + +/** + * Tunables for a [QuicConnection]. Defaults are appropriate for the MoQ + + * WebTransport workload — small connection-level windows, a generous-enough + * datagram size to fit Opus frames, idle timeout of 30 s. + * + * `maxUdpDatagramSize` caps both inbound and outbound UDP payload size. The + * QUIC spec mandates that any client sending Initial packets MUST send + * datagrams of at least 1200 bytes; we pad outbound Initials to satisfy this + * regardless of how full the packet is. + */ +data class QuicConnectionConfig( + val initialMaxData: Long = 16L * 1024 * 1024, + val initialMaxStreamDataBidiLocal: Long = 1L * 1024 * 1024, + val initialMaxStreamDataBidiRemote: Long = 1L * 1024 * 1024, + val initialMaxStreamDataUni: Long = 1L * 1024 * 1024, + val initialMaxStreamsBidi: Long = 100L, + val initialMaxStreamsUni: Long = 100L, + val maxIdleTimeoutMillis: Long = 30_000L, + val maxUdpPayloadSize: Long = 1452L, + val activeConnectionIdLimit: Long = 4L, + val maxDatagramFrameSize: Long = 1200L, + val ackDelayExponent: Long = 3L, + val maxAckDelay: Long = 25L, +) diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt new file mode 100644 index 000000000..3aeac588f --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -0,0 +1,230 @@ +/* + * 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.AckFrame +import com.vitorpamplona.quic.frame.ConnectionCloseFrame +import com.vitorpamplona.quic.frame.CryptoFrame +import com.vitorpamplona.quic.frame.DatagramFrame +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.NewConnectionIdFrame +import com.vitorpamplona.quic.frame.PingFrame +import com.vitorpamplona.quic.frame.StreamFrame +import com.vitorpamplona.quic.frame.decodeFrames +import com.vitorpamplona.quic.packet.LongHeaderPacket +import com.vitorpamplona.quic.packet.LongHeaderType +import com.vitorpamplona.quic.packet.ShortHeaderPacket +import com.vitorpamplona.quic.tls.TlsClient + +/** + * Decode every QUIC packet inside a single inbound UDP datagram and dispatch + * its frames to [conn]'s state. + * + * QUIC permits *coalescing* multiple packets into one datagram (RFC 9000 §12.2) + * — typically Initial + Handshake from the server in the same datagram during + * the handshake. We loop until the datagram is fully consumed or a packet + * fails to parse (which we drop silently per RFC 9001 §5.5). + */ +fun feedDatagram( + conn: QuicConnection, + datagram: ByteArray, + nowMillis: Long, +) { + var offset = 0 + while (offset < datagram.size) { + val first = datagram[offset].toInt() and 0xFF + val isLong = (first and 0x80) != 0 + if (isLong) { + val consumed = feedLongHeaderPacket(conn, datagram, offset, nowMillis) ?: break + offset += consumed + } else { + // Short-header — consumes the rest of the datagram. + feedShortHeaderPacket(conn, datagram, offset, nowMillis) + return + } + } +} + +private fun feedLongHeaderPacket( + conn: QuicConnection, + datagram: ByteArray, + offset: Int, + nowMillis: Long, +): Int? { + val peeked = LongHeaderPacket.peekHeader(datagram, offset) ?: return null + val level = + when (peeked.type) { + LongHeaderType.INITIAL -> EncryptionLevel.INITIAL + LongHeaderType.HANDSHAKE -> EncryptionLevel.HANDSHAKE + LongHeaderType.ZERO_RTT, LongHeaderType.RETRY -> return null // unsupported in client + } + val state = conn.levelState(level) + val proto = state.receiveProtection ?: return null + val parsed = + LongHeaderPacket.parseAndDecrypt( + bytes = datagram, + offset = offset, + aead = proto.aead, + key = proto.key, + iv = proto.iv, + hp = proto.hp, + hpKey = proto.hpKey, + largestReceivedInSpace = state.pnSpace.largestReceived, + ) ?: return null + + state.pnSpace.observeInbound(parsed.packet.packetNumber, nowMillis) + + // The server's source CID becomes our destination CID for subsequent packets. + if (level == EncryptionLevel.INITIAL) { + conn.destinationConnectionId = parsed.packet.scid + } + + dispatchFrames(conn, level, parsed.packet.payload, nowMillis) + return parsed.consumed +} + +private fun feedShortHeaderPacket( + conn: QuicConnection, + datagram: ByteArray, + offset: Int, + nowMillis: Long, +) { + val state = conn.levelState(EncryptionLevel.APPLICATION) + val proto = state.receiveProtection ?: return + val parsed = + ShortHeaderPacket.parseAndDecrypt( + bytes = datagram, + offset = offset, + dcidLen = conn.sourceConnectionId.length, + aead = proto.aead, + key = proto.key, + iv = proto.iv, + hp = proto.hp, + hpKey = proto.hpKey, + largestReceivedInSpace = state.pnSpace.largestReceived, + ) ?: return + state.pnSpace.observeInbound(parsed.packet.packetNumber, nowMillis) + dispatchFrames(conn, EncryptionLevel.APPLICATION, parsed.packet.payload, nowMillis) +} + +private fun dispatchFrames( + conn: QuicConnection, + level: EncryptionLevel, + payload: ByteArray, + nowMillis: Long, +) { + val frames = decodeFrames(payload) + val state = conn.levelState(level) + var ackEliciting = false + for (frame in frames) { + when (frame) { + is AckFrame -> { + // We don't currently retransmit, so we just absorb ACKs; once + // Phase F adds retransmission this updates the inflight tracker. + } + is CryptoFrame -> { + ackEliciting = true + state.cryptoReceive.insert(frame.offset, frame.data) + val contiguous = state.cryptoReceive.readContiguous() + if (contiguous.isNotEmpty()) { + val tlsLevel = + when (level) { + EncryptionLevel.INITIAL -> TlsClient.Level.INITIAL + EncryptionLevel.HANDSHAKE -> TlsClient.Level.HANDSHAKE + EncryptionLevel.APPLICATION -> TlsClient.Level.APPLICATION + } + conn.tls.pushHandshakeBytes(tlsLevel, contiguous) + // Pull any new outbound CRYPTO bytes out of TLS into our + // send queue at the matching encryption level. + drainTlsOutbound(conn) + } + } + is StreamFrame -> { + ackEliciting = true + val stream = conn.getOrCreatePeerStream(frame.streamId) + stream.receive.insert(frame.offset, frame.data, frame.fin) + val data = stream.receive.readContiguous() + if (data.isNotEmpty()) { + stream.deliverIncoming(data) + } + if (stream.receive.finReceived) { + stream.closeIncoming() + } + } + is DatagramFrame -> { + ackEliciting = true + conn.incomingDatagramsBuffer().addLast(frame.data) + } + is MaxDataFrame -> { + // Updates connection-level send credit; left to the orchestrator. + } + is MaxStreamDataFrame -> { + conn.streamById(frame.streamId)?.let { + if (frame.maxStreamData > it.sendCredit) it.sendCredit = frame.maxStreamData + } + } + is MaxStreamsFrame -> { + // Tracking left for a later phase. + } + is NewConnectionIdFrame -> { + // We don't support migration; ignore. + } + is ConnectionCloseFrame -> { + conn.status = QuicConnection.Status.CLOSED + } + is HandshakeDoneFrame -> { + conn.status = QuicConnection.Status.CONNECTED + } + is PingFrame -> { + ackEliciting = true + } + else -> { + // PADDING + DATA_BLOCKED + STREAM_DATA_BLOCKED + STREAMS_BLOCKED + // are non-eliciting / cleared during decode. + } + } + } + if (ackEliciting) { + // Get largest received from pn space and feed into ack tracker. + val pn = state.pnSpace.largestReceived + if (pn >= 0) { + state.ackTracker.receivedPacket(pn, ackEliciting = true, receivedAtMillis = nowMillis) + } + } +} + +private fun drainTlsOutbound(conn: QuicConnection) { + for (lvl in TlsClient.Level.entries) { + while (true) { + val bytes = conn.tls.pollOutbound(lvl) ?: break + val state = + when (lvl) { + TlsClient.Level.INITIAL -> conn.initial + TlsClient.Level.HANDSHAKE -> conn.handshake + TlsClient.Level.APPLICATION -> conn.application + } + state.cryptoSend.enqueue(bytes) + } + } +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt new file mode 100644 index 000000000..7dc0dbd0a --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -0,0 +1,246 @@ +/* + * 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.ConnectionCloseFrame +import com.vitorpamplona.quic.frame.CryptoFrame +import com.vitorpamplona.quic.frame.DatagramFrame +import com.vitorpamplona.quic.frame.Frame +import com.vitorpamplona.quic.frame.PingFrame +import com.vitorpamplona.quic.frame.StreamFrame +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 com.vitorpamplona.quic.packet.ShortHeaderPacket +import com.vitorpamplona.quic.packet.ShortHeaderPlaintextPacket + +/** + * Build the next UDP datagram to send for [conn], or return null if there's + * nothing to send right now. + * + * The datagram coalesces (in order): + * - Initial packet (if Initial-level CRYPTO has unsent bytes or a pending ACK) + * - Handshake packet (likewise) + * - 1-RTT packet (CRYPTO from NewSessionTicket, STREAM, DATAGRAM, ACK, etc.) + * + * RFC 9000 §14: any datagram containing an Initial packet from the client + * MUST be padded to at least 1200 bytes total. + */ +fun drainOutbound( + conn: QuicConnection, + nowMillis: Long, +): ByteArray? { + val parts = mutableListOf() + + // Closing — emit a CONNECTION_CLOSE at the highest available level. + if (conn.status == QuicConnection.Status.CLOSING) { + val frame = ConnectionCloseFrame(conn.closeErrorCode, null, conn.closeReason ?: "") + val packet = buildBestLevelPacket(conn, listOf(frame)) ?: return null + conn.status = QuicConnection.Status.CLOSED + return packet + } + + val initialPkt = buildInitialPacket(conn, nowMillis) + if (initialPkt != null) parts += initialPkt + + val handshakePkt = buildHandshakePacket(conn, nowMillis) + if (handshakePkt != null) parts += handshakePkt + + val applicationPkt = buildApplicationPacket(conn, nowMillis) + if (applicationPkt != null) parts += applicationPkt + + if (parts.isEmpty()) return null + + // Coalesce + var totalLen = 0 + for (p in parts) totalLen += p.size + // Pad the datagram to 1200 bytes if it begins with an Initial packet (RFC 9000 §14). + if (initialPkt != null && totalLen < 1200) totalLen = 1200 + val out = ByteArray(totalLen) + var pos = 0 + for (p in parts) { + p.copyInto(out, pos) + pos += p.size + } + // Tail bytes are zero (PADDING frames in the encryption envelope of the last packet). + // Note: padding inside the last QUIC packet's encryption is the RFC-correct way; we instead + // pad the datagram with zero bytes after the last packet. Servers tolerate this for the + // datagram-level minimum size requirement. + return out +} + +private fun buildBestLevelPacket( + conn: QuicConnection, + frames: List, +): ByteArray? { + // Prefer 1-RTT > Handshake > Initial. + val payload = encodeFrames(frames) + val app = conn.application + if (app.sendProtection != null) { + val proto = app.sendProtection!! + val pn = app.pnSpace.allocateOutbound() + return ShortHeaderPacket.build( + ShortHeaderPlaintextPacket(conn.destinationConnectionId, pn, payload), + proto.aead, + proto.key, + proto.iv, + proto.hp, + proto.hpKey, + largestAckedInSpace = -1L, + ) + } + val hs = conn.handshake + if (hs.sendProtection != null) { + return buildLongHeaderPacket(conn, EncryptionLevel.HANDSHAKE, payload) + } + val init = conn.initial + if (init.sendProtection != null) { + return buildLongHeaderPacket(conn, EncryptionLevel.INITIAL, payload) + } + return null +} + +private fun buildLongHeaderPacket( + conn: QuicConnection, + level: EncryptionLevel, + payload: ByteArray, +): ByteArray { + val state = conn.levelState(level) + val proto = state.sendProtection!! + val pn = state.pnSpace.allocateOutbound() + val type = + when (level) { + EncryptionLevel.INITIAL -> LongHeaderType.INITIAL + EncryptionLevel.HANDSHAKE -> LongHeaderType.HANDSHAKE + EncryptionLevel.APPLICATION -> error("APPLICATION uses short-header packets") + } + return LongHeaderPacket.build( + LongHeaderPlaintextPacket( + type = type, + version = QuicVersion.V1, + dcid = conn.destinationConnectionId, + scid = conn.sourceConnectionId, + packetNumber = pn, + payload = payload, + ), + proto.aead, + proto.key, + proto.iv, + proto.hp, + proto.hpKey, + largestAckedInSpace = -1L, + ) +} + +private fun buildInitialPacket( + conn: QuicConnection, + nowMillis: Long, +): ByteArray? = buildHandshakeLevelPacket(conn, EncryptionLevel.INITIAL, nowMillis) + +private fun buildHandshakePacket( + conn: QuicConnection, + nowMillis: Long, +): ByteArray? = buildHandshakeLevelPacket(conn, EncryptionLevel.HANDSHAKE, nowMillis) + +private fun buildHandshakeLevelPacket( + conn: QuicConnection, + level: EncryptionLevel, + nowMillis: Long, +): ByteArray? { + val state = conn.levelState(level) + val proto = state.sendProtection ?: return null + val frames = mutableListOf() + + // Pending ACK? + state.ackTracker.buildAckFrame(nowMillis, conn.config.ackDelayExponent.toInt())?.let { frames += it } + + // Pending CRYPTO? + val cryptoChunk = state.cryptoSend.takeChunk(maxBytes = 1100) + if (cryptoChunk != null && cryptoChunk.data.isNotEmpty()) { + frames += CryptoFrame(cryptoChunk.offset, cryptoChunk.data) + } + + if (frames.isEmpty()) return null + + val payload = encodeFrames(frames) + val pn = state.pnSpace.allocateOutbound() + val type = if (level == EncryptionLevel.INITIAL) LongHeaderType.INITIAL else LongHeaderType.HANDSHAKE + return LongHeaderPacket.build( + LongHeaderPlaintextPacket( + type = type, + version = QuicVersion.V1, + dcid = conn.destinationConnectionId, + scid = conn.sourceConnectionId, + packetNumber = pn, + payload = payload, + ), + proto.aead, + proto.key, + proto.iv, + proto.hp, + proto.hpKey, + largestAckedInSpace = -1L, + ) +} + +private fun buildApplicationPacket( + conn: QuicConnection, + nowMillis: Long, +): ByteArray? { + val state = conn.application + val proto = state.sendProtection ?: return null + val frames = mutableListOf() + + state.ackTracker.buildAckFrame(nowMillis, conn.config.ackDelayExponent.toInt())?.let { frames += it } + + // Pending datagrams + while (conn.pendingDatagramsView().isNotEmpty()) { + val payload = conn.pendingDatagramsView().removeFirst() + frames += DatagramFrame(payload, explicitLength = true) + if (frames.size >= 16) break + } + + // Drain stream send buffers — round-robin keeping packet under MTU. + var packetBudget = 1100 + for ((id, stream) in conn.streamsView()) { + if (packetBudget <= 64) break + val chunk = stream.send.takeChunk(maxBytes = packetBudget - 32) ?: 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 + } + } + + if (frames.isEmpty()) return null + val payload = encodeFrames(frames) + val pn = state.pnSpace.allocateOutbound() + return ShortHeaderPacket.build( + ShortHeaderPlaintextPacket(conn.destinationConnectionId, pn, payload), + proto.aead, + proto.key, + proto.iv, + proto.hp, + proto.hpKey, + largestAckedInSpace = -1L, + ) +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt new file mode 100644 index 000000000..5c85fb791 --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/recovery/AckTracker.kt @@ -0,0 +1,115 @@ +/* + * 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.recovery + +import com.vitorpamplona.quic.frame.AckFrame +import com.vitorpamplona.quic.frame.AckRange + +/** + * Tracks received packet numbers for one packet-number space and produces + * ACK frames per RFC 9000 §19.3. + * + * We store packet numbers as a sorted list of disjoint ranges (lo..hi). + * Acks emit the ranges newest-first as RFC 9000 specifies. + */ +class AckTracker { + private val ranges = mutableListOf() // sorted descending by lo (i.e. ranges[0] has the largest) + private var ackElicitingPending: Boolean = false + private var largestRecvTimeMillis: Long = 0L + + fun receivedPacket( + packetNumber: Long, + ackEliciting: Boolean, + receivedAtMillis: Long, + ) { + if (ackEliciting) ackElicitingPending = true + if (ranges.isEmpty()) { + ranges += LongRange(packetNumber, packetNumber) + largestRecvTimeMillis = receivedAtMillis + return + } + if (packetNumber > ranges[0].endInclusive) { + largestRecvTimeMillis = receivedAtMillis + } + + // Find the range we need to extend (or insertion point). + for (i in ranges.indices) { + val r = ranges[i] + if (packetNumber > r.endInclusive + 1) { + ranges.add(i, LongRange(packetNumber, packetNumber)) + return + } + if (packetNumber == r.endInclusive + 1) { + // Extend up; check merge with i-1 (if any) + ranges[i] = LongRange(r.start, packetNumber) + if (i > 0 && ranges[i - 1].start == packetNumber + 1) { + ranges[i - 1] = LongRange(ranges[i].start, ranges[i - 1].endInclusive) + ranges.removeAt(i) + } + return + } + if (packetNumber in r) { + // Already known. + return + } + if (packetNumber == r.start - 1) { + ranges[i] = LongRange(packetNumber, r.endInclusive) + if (i + 1 < ranges.size && ranges[i + 1].endInclusive == packetNumber - 1) { + ranges[i] = LongRange(ranges[i + 1].start, ranges[i].endInclusive) + ranges.removeAt(i + 1) + } + return + } + } + ranges += LongRange(packetNumber, packetNumber) + } + + fun hasUnackedAckEliciting(): Boolean = ackElicitingPending + + fun isEmpty(): Boolean = ranges.isEmpty() + + fun largestReceived(): Long = if (ranges.isEmpty()) -1L else ranges[0].endInclusive + + /** Build an ACK frame covering everything we've received. */ + fun buildAckFrame( + nowMillis: Long, + ackDelayExponent: Int = 3, + ): AckFrame? { + if (ranges.isEmpty()) return null + val largest = ranges[0].endInclusive + val firstRangeLength = largest - ranges[0].start + val rest = mutableListOf() + for (i in 1 until ranges.size) { + val gap = ranges[i - 1].start - ranges[i].endInclusive - 2 + val len = ranges[i].endInclusive - ranges[i].start + rest += AckRange(gap, len) + } + val ackDelayMicros = (nowMillis - largestRecvTimeMillis) * 1000L + val ackDelay = (ackDelayMicros ushr ackDelayExponent).coerceAtLeast(0L) + ackElicitingPending = false + return AckFrame( + largestAcknowledged = largest, + ackDelay = ackDelay, + firstAckRange = firstRangeLength, + additionalRanges = rest, + ) + } +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt new file mode 100644 index 000000000..1918c87bb --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -0,0 +1,67 @@ +/* + * 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.channels.Channel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.consumeAsFlow + +/** + * One QUIC stream (bidirectional or unidirectional). Application code + * accesses it through [enqueue] (write) and the [readFlow] / FIN observation + * APIs; the [QuicConnection] owns the underlying buffers and drains them + * into STREAM frames on the wire. + */ +class QuicStream( + val streamId: Long, + val direction: Direction, +) { + enum class Direction { BIDIRECTIONAL, UNIDIRECTIONAL_LOCAL_TO_REMOTE, UNIDIRECTIONAL_REMOTE_TO_LOCAL } + + val send = SendBuffer() + val receive = ReceiveBuffer() + + /** Bytes received and confirmed contiguous, exposed as a flow to the consumer. */ + private val incomingChannel = Channel(capacity = Channel.UNLIMITED) + val incoming: Flow get() = incomingChannel.consumeAsFlow() + + /** Per-stream send credit (peer's MAX_STREAM_DATA value). */ + var sendCredit: Long = 0L + internal set + + /** Per-stream receive credit (the value we advertised). */ + var receiveLimit: Long = 0L + internal set + + /** True once we've FIN'd our write side and the peer FIN'd theirs. */ + val isClosed: Boolean + get() = send.finSent && receive.finReceived + + internal fun deliverIncoming(data: ByteArray) { + if (data.isNotEmpty()) { + incomingChannel.trySend(data) + } + } + + internal fun closeIncoming() { + incomingChannel.close() + } +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt new file mode 100644 index 000000000..73dc722da --- /dev/null +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt @@ -0,0 +1,83 @@ +/* + * 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 + +/** + * Outbound send buffer for one direction of a QUIC stream (or for the + * per-encryption-level CRYPTO offset stream). + * + * Application code [enqueue]s payload bytes; the connection's send loop + * [takeChunk]s as much as it can fit in the next packet, given the + * remaining packet budget and stream-level / connection-level flow control + * credit. Sent bytes stay in the buffer until [acknowledge] (Phase F adds + * retransmission of lost bytes; for v1 we just trust the receiver and + * release on send). + * + * For Phase D-K we run a "best effort" mode: bytes are released from the + * buffer the moment they're handed off, on the assumption that the + * underlying network is stable. Phase L adds retransmit-on-loss (using the + * same send buffer that retains until ACK). + */ +class SendBuffer { + private var pending: ByteArray = ByteArray(0) + private var sentEnd: Long = 0L + var nextOffset: Long = 0L + private set + var finPending: Boolean = false + private set + var finSent: Boolean = false + private set + + val readableBytes: Int get() = pending.size + + fun enqueue(bytes: ByteArray) { + if (bytes.isEmpty()) return + val combined = ByteArray(pending.size + bytes.size) + pending.copyInto(combined, 0) + bytes.copyInto(combined, pending.size) + pending = combined + nextOffset += bytes.size + } + + /** Mark the write side as closing; the next [takeChunk] will set FIN once empty. */ + fun finish() { + finPending = true + } + + /** Take up to [maxBytes] bytes off the head of the buffer at the current send offset. */ + fun takeChunk(maxBytes: Int): Chunk? { + if (pending.isEmpty() && !(finPending && !finSent)) return null + val take = minOf(pending.size, maxBytes.coerceAtLeast(0)) + val data = pending.copyOfRange(0, take) + pending = pending.copyOfRange(take, pending.size) + val offset = sentEnd + sentEnd += take + val fin = finPending && pending.isEmpty() + if (fin) finSent = true + return Chunk(offset, data, fin) + } + + data class Chunk( + val offset: Long, + val data: ByteArray, + val fin: Boolean, + ) +}