feat(quic): emit RESET_STREAM / STOP_SENDING + per-stream retransmit dispatch

Application code can now call QuicStream.resetStream(errorCode) and
stopSending(errorCode); the next writer drain emits the matching frame
with a RecoveryToken. Loss dispatch re-flags the per-stream emit-pending
bit; ACK dispatch latches resetAcked / stopSendingAcked so stale loss
notifications can't re-emit. NEW_CONNECTION_ID retransmit drains
QuicConnection.pendingNewConnectionId on next writer pass (no public
emit API since :quic doesn't rotate connection IDs, but the wiring is
in place for a future emit path).

Five tests in ResetStopSendingEmitTest mirror neqo's send-stream reset
coverage: emit-and-token, retransmit-on-loss, ack-then-stale-loss-drop,
stop-sending emission, and NEW_CONNECTION_ID drain.

https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ

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

@@ -205,27 +205,14 @@ class QuicConnection(
      */
     internal val pendingMaxStreamData: MutableMap<Long, Long> = 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<Long, com.vitorpamplona.quic.connection.recovery.RecoveryToken.ResetStream> = HashMap()
-
-    /**
-     * Pending retransmits for `STOP_SENDING` frames (RFC 9000 §19.5).
-     * Same scaffolding-only status as [pendingResetStream].
-     */
-    internal val pendingStopSending:
-        MutableMap<Long, com.vitorpamplona.quic.connection.recovery.RecoveryToken.StopSending> = HashMap()
-
     /**
      * Pending retransmits for `NEW_CONNECTION_ID` frames (RFC 9000
-     * §19.15). Keyed by sequenceNumber. Same scaffolding-only status
+     * §19.15). Keyed by sequence number. Populated on loss via
-     * as [pendingResetStream].
+     * [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<Long, com.vitorpamplona.quic.connection.recovery.RecoveryToken.NewConnectionId> = 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
+                    // Flow-control extensions and NEW_CONNECTION_ID
-                    // ACK. (Pending* maps are populated only on loss; an ACK
+                    // have no per-buffer state to release on ACK; the
-                    // for a frame that never lost is naturally absent.)
+                    // 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
+                    // Reliable per RFC 9000 §13.3. Re-flag the
-                    // on loss. `:quic` doesn't currently have an emit
+                    // owning stream's RESET_STREAM emit pending so
-                    // path for RESET_STREAM, so the pending-list is
+                    // the next writer drain re-emits — unless the
-                    // populated but never drained today; the field is
+                    // peer already ACK'd it (resetAcked == true), in
-                    // scaffolding for future emit support.
+                    // which case the lost token is a stale duplicate
-                    pendingResetStream[token.streamId] = token
+                    // 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 -> {

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

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,
+    )
 }

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<RecoveryToken.ResetStream>()
+                    .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<RecoveryToken.ResetStream>()
+                    .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<RecoveryToken.ResetStream>()
+                    .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<RecoveryToken.ResetStream>()
+                    .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<RecoveryToken.StopSending>()
+                    .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
+        }
+}