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 c3d5b6b3c..171533a62 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -205,27 +205,14 @@ class QuicConnection( */ internal val pendingMaxStreamData: MutableMap = HashMap() - /** - * Pending retransmits for `RESET_STREAM` frames (RFC 9000 §19.4). - * Keyed by stream id; the dispatcher records lost RESET_STREAM - * tokens here. `:quic` has no emit path for RESET_STREAM today, - * so the writer never drains this map — scaffolding for future - * stream-reset support. Caller must hold [lock]. - */ - internal val pendingResetStream: - MutableMap = HashMap() - - /** - * Pending retransmits for `STOP_SENDING` frames (RFC 9000 §19.5). - * Same scaffolding-only status as [pendingResetStream]. - */ - internal val pendingStopSending: - MutableMap = HashMap() - /** * Pending retransmits for `NEW_CONNECTION_ID` frames (RFC 9000 - * §19.15). Keyed by sequenceNumber. Same scaffolding-only status - * as [pendingResetStream]. + * §19.15). Keyed by sequence number. Populated on loss via + * [onTokensLost]; drained by the writer on the next outbound. + * Connection-ID rotation isn't currently triggered by any + * application path inside `:quic`, so the public emit API is + * limited to test usage — but the retransmit machinery is + * complete so any future emit path lands cleanly. */ internal val pendingNewConnectionId: MutableMap = HashMap() @@ -790,13 +777,28 @@ class QuicConnection( is com.vitorpamplona.quic.connection.recovery.RecoveryToken.MaxStreamsBidi, is com.vitorpamplona.quic.connection.recovery.RecoveryToken.MaxData, is com.vitorpamplona.quic.connection.recovery.RecoveryToken.MaxStreamData, - is com.vitorpamplona.quic.connection.recovery.RecoveryToken.ResetStream, - is com.vitorpamplona.quic.connection.recovery.RecoveryToken.StopSending, is com.vitorpamplona.quic.connection.recovery.RecoveryToken.NewConnectionId, -> { - // Control frames have no per-buffer state to release on - // ACK. (Pending* maps are populated only on loss; an ACK - // for a frame that never lost is naturally absent.) + // Flow-control extensions and NEW_CONNECTION_ID + // have no per-buffer state to release on ACK; the + // pending* maps are populated only on loss, so an + // ACK for a frame that never lost is naturally + // absent. + } + + is com.vitorpamplona.quic.connection.recovery.RecoveryToken.ResetStream -> { + // Latch resetAcked = true. Subsequent loss + // notifications for stale RESET_STREAM tokens are + // dropped (see onTokensLost). + val stream = streamByIdLocked(token.streamId) ?: continue + stream.resetAcked = true + stream.resetEmitPending = false + } + + is com.vitorpamplona.quic.connection.recovery.RecoveryToken.StopSending -> { + val stream = streamByIdLocked(token.streamId) ?: continue + stream.stopSendingAcked = true + stream.stopSendingEmitPending = false } is com.vitorpamplona.quic.connection.recovery.RecoveryToken.Stream -> { @@ -888,16 +890,19 @@ class QuicConnection( } is com.vitorpamplona.quic.connection.recovery.RecoveryToken.ResetStream -> { - // Reliable per RFC 9000 §13.3 — re-emit verbatim - // on loss. `:quic` doesn't currently have an emit - // path for RESET_STREAM, so the pending-list is - // populated but never drained today; the field is - // scaffolding for future emit support. - pendingResetStream[token.streamId] = token + // Reliable per RFC 9000 §13.3. Re-flag the + // owning stream's RESET_STREAM emit pending so + // the next writer drain re-emits — unless the + // peer already ACK'd it (resetAcked == true), in + // which case the lost token is a stale duplicate + // and we drop it. + val stream = streamByIdLocked(token.streamId) ?: continue + if (!stream.resetAcked) stream.resetEmitPending = true } is com.vitorpamplona.quic.connection.recovery.RecoveryToken.StopSending -> { - pendingStopSending[token.streamId] = token + val stream = streamByIdLocked(token.streamId) ?: continue + if (!stream.stopSendingAcked) stream.stopSendingEmitPending = true } is com.vitorpamplona.quic.connection.recovery.RecoveryToken.NewConnectionId -> { 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 615eb0a5e..86560fb56 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -31,7 +31,10 @@ import com.vitorpamplona.quic.frame.Frame 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.ResetStreamFrame +import com.vitorpamplona.quic.frame.StopSendingFrame import com.vitorpamplona.quic.frame.StreamFrame import com.vitorpamplona.quic.frame.encodeFrames import com.vitorpamplona.quic.packet.LongHeaderPacket @@ -535,6 +538,65 @@ private fun appendFlowControlUpdates( } conn.pendingMaxStreamData.clear() } + + // Per-stream RESET_STREAM / STOP_SENDING. Both are reliable per + // RFC 9000 §13.3 — the writer drains the per-stream emit-pending + // flag, attaches the corresponding RecoveryToken, and clears the + // flag. The loss dispatcher re-flags on packet loss; ACK latches + // resetAcked / stopSendingAcked so subsequent stale loss tokens + // are dropped. + for (stream in conn.streamsListLocked()) { + val resetState = stream.resetState + if (resetState != null && stream.resetEmitPending && !stream.resetAcked) { + frames += + ResetStreamFrame( + streamId = stream.streamId, + applicationErrorCode = resetState.errorCode, + finalSize = resetState.finalSize, + ) + tokens += + RecoveryToken.ResetStream( + streamId = stream.streamId, + errorCode = resetState.errorCode, + finalSize = resetState.finalSize, + ) + stream.resetEmitPending = false + } + val stopSendingState = stream.stopSendingState + if (stopSendingState != null && stream.stopSendingEmitPending && !stream.stopSendingAcked) { + frames += + StopSendingFrame( + streamId = stream.streamId, + applicationErrorCode = stopSendingState.errorCode, + ) + tokens += + RecoveryToken.StopSending( + streamId = stream.streamId, + errorCode = stopSendingState.errorCode, + ) + stream.stopSendingEmitPending = false + } + } + + // NEW_CONNECTION_ID retransmits. No application path emits these + // initially today (connection-ID rotation isn't wired); the map + // is populated only by the loss dispatcher, so this branch only + // fires for genuine retransmits. + if (conn.pendingNewConnectionId.isNotEmpty()) { + val pendingNewCidEntries = conn.pendingNewConnectionId.entries.toList() + for ((_, token) in pendingNewCidEntries) { + frames += + NewConnectionIdFrame( + sequenceNumber = token.sequenceNumber, + retirePriorTo = token.retirePriorTo, + connectionId = token.connectionId, + statelessResetToken = token.statelessResetToken, + ) + tokens += token + } + conn.pendingNewConnectionId.clear() + } + val cfg = conn.config var totalRecvAdvanced = 0L // Round-4 perf #9 + round-5 #9: walk the streams via the index-friendly diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt index e9c29c985..5e5cfd6a2 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -100,4 +100,81 @@ class QuicStream( internal fun closeIncoming() { incomingChannel.close() } + + /** + * RFC 9000 §3.5: abruptly terminate the SEND side. Application code + * calls this when it wants to cancel a partially-written stream + * (e.g. user cancelled a request mid-upload). After [resetStream]: + * + * - A `RESET_STREAM(streamId, errorCode, finalSize)` frame is + * queued for emission on the next writer drain. `finalSize` is + * the largest stream offset the application has appended, i.e. + * [SendBuffer.nextOffset] at reset time (RFC 9000 §3.5 — the + * peer uses this to satisfy its receive-side flow-control + * accounting even though it discards the bytes). + * - The frame is reliable per §13.3 — if its carrying packet is + * declared lost, the dispatcher in + * [com.vitorpamplona.quic.connection.QuicConnection.onTokensLost] + * re-flags [resetEmitPending] for re-emit on next drain. + * - Once the peer ACKs the RESET_STREAM, [resetAcked] latches + * true and the writer stops re-emitting. + * + * Idempotent: a second [resetStream] call overwrites the queued + * entry with the newer error code (in practice apps shouldn't + * reset twice — the second call is benign). + */ + fun resetStream(errorCode: Long) { + resetState = + ResetState( + errorCode = errorCode, + finalSize = send.nextOffset, + ) + resetEmitPending = true + } + + /** + * RFC 9000 §3.5: ask the peer to stop sending on this stream's + * RECEIVE side. Application calls this when it's done reading + * (e.g. an HTTP/3 request handler returned an early response and + * doesn't care about the rest of the request body). The + * `STOP_SENDING(streamId, errorCode)` frame is queued for emission + * on the next writer drain; reliable per §13.3, retransmitted on + * loss like [resetStream]. + */ + fun stopSending(errorCode: Long) { + stopSendingState = StopSendingState(errorCode = errorCode) + stopSendingEmitPending = true + } + + /** RFC 9000 §3.5 send-side reset state. Set by [resetStream]. */ + internal var resetState: ResetState? = null + + /** + * True while a RESET_STREAM emit is pending. Cleared after the + * writer emits; set back to true by the loss dispatcher; cleared + * again (and not re-set) once [resetAcked] latches. + */ + internal var resetEmitPending: Boolean = false + + /** + * Latches true when the peer ACKs the RESET_STREAM. After this + * the writer no longer re-emits even if a stale lost token shows + * up. Mirrors neqo's `send_stream::ResetAcked` state. + */ + internal var resetAcked: Boolean = false + + /** Receive-side stop-sending state. Set by [stopSending]. */ + internal var stopSendingState: StopSendingState? = null + + internal var stopSendingEmitPending: Boolean = false + internal var stopSendingAcked: Boolean = false + + internal data class ResetState( + val errorCode: Long, + val finalSize: Long, + ) + + internal data class StopSendingState( + val errorCode: Long, + ) } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt new file mode 100644 index 000000000..849c3bd00 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt @@ -0,0 +1,258 @@ +/* + * 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.connection.recovery.RecoveryToken +import com.vitorpamplona.quic.tls.InProcessTlsServer +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertTrue + +/** + * Public-API + writer drain for RESET_STREAM / STOP_SENDING / + * NEW_CONNECTION_ID. Verifies: + * + * 1. App calls [com.vitorpamplona.quic.stream.QuicStream.resetStream] + * / [com.vitorpamplona.quic.stream.QuicStream.stopSending]; the + * next writer drain emits the matching frame with a token. + * 2. Loss dispatch re-flags the per-stream emit-pending bit; next + * drain re-emits. + * 3. ACK dispatch latches resetAcked / stopSendingAcked; subsequent + * stale loss tokens are dropped. + * 4. NEW_CONNECTION_ID retransmit drains + * [QuicConnection.pendingNewConnectionId]. + */ +class ResetStopSendingEmitTest { + @Test + fun resetStream_emitsResetStreamFrameAndToken() = + runBlocking { + val client = handshakedClient() + val stream = client.openUniStream() + stream.send.enqueue("partial-upload".encodeToByteArray()) + // App cancels mid-write. + stream.resetStream(errorCode = 7L) + + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + val tokenEntry = + newEntries.firstOrNull { entry -> + entry.value.tokens.any { it is RecoveryToken.ResetStream } + } + assertNotNull(tokenEntry, "resetStream() must produce a SentPacket carrying ResetStream") + val token = + tokenEntry.value.tokens + .filterIsInstance() + .single() + assertEquals(stream.streamId, token.streamId) + assertEquals(7L, token.errorCode) + // finalSize == nextOffset at reset time. We enqueued 14 bytes. + assertEquals(14L, token.finalSize, "finalSize is the largest enqueued offset") + // Pending flag cleared after emit. + assertEquals(false, stream.resetEmitPending) + assertEquals(false, stream.resetAcked, "ACK hasn't arrived yet") + } + + @Test + fun resetStream_retransmittedOnLoss() = + runBlocking { + val client = handshakedClient() + val stream = client.openUniStream() + stream.resetStream(errorCode = 13L) + runCatching { drainOutbound(client, nowMillis = 1L) } + + val firstEntry = + client.application.sentPackets.entries + .first { it.value.tokens.any { it is RecoveryToken.ResetStream } } + val token = + firstEntry.value.tokens + .filterIsInstance() + .single() + + // Simulate loss. + client.lock.lock() + try { + client.onTokensLost(listOf(token)) + client.application.sentPackets.remove(firstEntry.key) + } finally { + client.lock.unlock() + } + // Per-stream emit-pending should be re-flagged. + assertTrue(stream.resetEmitPending, "loss must re-flag resetEmitPending") + + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 2L) } + val replay = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + .firstOrNull { entry -> entry.value.tokens.any { it is RecoveryToken.ResetStream } } + assertNotNull(replay, "retransmit must produce a fresh SentPacket carrying ResetStream") + val replayToken = + replay.value.tokens + .filterIsInstance() + .single() + assertEquals(token, replayToken, "retransmit replays identical RESET_STREAM contents") + } + + @Test + fun resetStream_acked_doesNotReEmitOnStaleLoss() = + runBlocking { + // ACK comes first, then a stale loss token arrives. The + // dispatcher must NOT re-flag resetEmitPending. + val client = handshakedClient() + val stream = client.openUniStream() + stream.resetStream(errorCode = 99L) + runCatching { drainOutbound(client, nowMillis = 1L) } + + val packet = + client.application.sentPackets.entries + .first { it.value.tokens.any { it is RecoveryToken.ResetStream } } + val token = + packet.value.tokens + .filterIsInstance() + .single() + + // ACK first. + client.lock.lock() + try { + client.onTokensAcked(listOf(token)) + } finally { + client.lock.unlock() + } + assertEquals(true, stream.resetAcked) + assertEquals(false, stream.resetEmitPending) + + // Now a stale loss notification arrives. Defensive: drop. + client.lock.lock() + try { + client.onTokensLost(listOf(token)) + } finally { + client.lock.unlock() + } + assertEquals(false, stream.resetEmitPending, "stale loss after ACK must not re-flag emit-pending") + } + + @Test + fun stopSending_emitsStopSendingFrameAndToken() = + runBlocking { + val client = handshakedClient() + val stream = client.openBidiStream() + stream.stopSending(errorCode = 5L) + + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val newEntries = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + val tokenEntry = + newEntries.firstOrNull { entry -> + entry.value.tokens.any { it is RecoveryToken.StopSending } + } + assertNotNull(tokenEntry) + val token = + tokenEntry.value.tokens + .filterIsInstance() + .single() + assertEquals(stream.streamId, token.streamId) + assertEquals(5L, token.errorCode) + assertEquals(false, stream.stopSendingEmitPending) + } + + @Test + fun newConnectionId_retransmittedOnLoss() = + runBlocking { + val client = handshakedClient() + // Inject a NewConnectionId loss token directly — there's no + // public emit API for NEW_CONNECTION_ID since :quic doesn't + // do connection-ID rotation, but the retransmit machinery + // is wired so a future emit path lands cleanly. + val token = + RecoveryToken.NewConnectionId( + sequenceNumber = 1L, + retirePriorTo = 0L, + connectionId = byteArrayOf(1, 2, 3, 4), + statelessResetToken = ByteArray(16) { it.toByte() }, + ) + client.lock.lock() + try { + client.onTokensLost(listOf(token)) + } finally { + client.lock.unlock() + } + assertEquals(token, client.pendingNewConnectionId[1L]) + + // Next drain emits the NEW_CONNECTION_ID frame and clears the map. + val sizeBefore = client.application.sentPackets.size + runCatching { drainOutbound(client, nowMillis = 1L) } + val replay = + client.application.sentPackets.entries + .sortedBy { it.key } + .drop(sizeBefore) + .firstOrNull { entry -> entry.value.tokens.any { it is RecoveryToken.NewConnectionId } } + assertNotNull(replay, "writer must drain pendingNewConnectionId") + assertTrue(client.pendingNewConnectionId.isEmpty(), "map must be cleared after drain") + } + + private fun handshakedClient(): QuicConnection = + runBlocking { + val client = + QuicConnection( + serverName = "example.test", + config = QuicConnectionConfig(), + tlsCertificateValidator = + com.vitorpamplona.quic.tls + .PermissiveCertificateValidator(), + ) + val serverScid = ConnectionId.random(8) + val tlsServer = + InProcessTlsServer( + transportParameters = + TransportParameters( + initialMaxData = 1_000_000, + initialMaxStreamDataBidiLocal = 100_000, + initialMaxStreamDataBidiRemote = 100_000, + initialMaxStreamDataUni = 100_000, + initialMaxStreamsBidi = 100, + initialMaxStreamsUni = 100, + initialSourceConnectionId = serverScid.bytes, + originalDestinationConnectionId = client.destinationConnectionId.bytes, + ).encode(), + ) + val pipe = + InMemoryQuicPipe( + client = client, + initialDcid = client.destinationConnectionId.bytes, + serverScid = serverScid, + tlsServer = tlsServer, + ) + client.start() + pipe.drive(maxRounds = 16) + assertEquals(QuicConnection.Status.CONNECTED, client.status) + client + } +}