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 5e5cfd6a2..1297250d8 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -119,11 +119,21 @@ class QuicStream( * - 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). + * First call wins. A second [resetStream] is a no-op: RFC 9000 §3.5 + * pins `finalSize` at the first emission; replaying retransmits with + * a larger value (because the app enqueued more bytes between the + * two calls) would trigger `FINAL_SIZE_ERROR` on the peer. The + * `errorCode` is likewise frozen. + * + * Threading: callable from any coroutine. The boolean flags + * ([resetEmitPending], [resetAcked]) are `@Volatile` so the writer + * (which reads them under [com.vitorpamplona.quic.connection.QuicConnection.lock]) + * sees the assignment without acquiring the connection mutex. The + * "first call wins" gate above closes the only multi-writer race + * (two app threads racing the writer's clear-after-emit). */ fun resetStream(errorCode: Long) { + if (resetState != null) return resetState = ResetState( errorCode = errorCode, @@ -140,33 +150,56 @@ class QuicStream( * `STOP_SENDING(streamId, errorCode)` frame is queued for emission * on the next writer drain; reliable per §13.3, retransmitted on * loss like [resetStream]. + * + * First call wins (same reasoning as [resetStream]: the peer treats + * the first STOP_SENDING as authoritative; replaying retransmits + * with a different `errorCode` would visibly disagree with the + * original frame already on the wire). */ fun stopSending(errorCode: Long) { + if (stopSendingState != null) return stopSendingState = StopSendingState(errorCode = errorCode) stopSendingEmitPending = true } - /** RFC 9000 §3.5 send-side reset state. Set by [resetStream]. */ + /** + * RFC 9000 §3.5 send-side reset state. Set once by [resetStream] + * (subsequent calls no-op); read by the writer + loss/ACK + * dispatchers. Once set, contents are immutable. + */ 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. + * + * `@Volatile` because [resetStream] writes it from an arbitrary + * coroutine while the writer / dispatchers read it under + * [com.vitorpamplona.quic.connection.QuicConnection.lock]. Volatile + * gives the cross-thread happens-before; the "first call wins" + * gate in [resetStream] eliminates the write-write race with the + * writer's clear-after-emit. */ + @Volatile 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. + * up. Mirrors neqo's `send_stream::ResetAcked` state. Volatile for + * the same reason as [resetEmitPending]. */ + @Volatile internal var resetAcked: Boolean = false /** Receive-side stop-sending state. Set by [stopSending]. */ internal var stopSendingState: StopSendingState? = null + @Volatile internal var stopSendingEmitPending: Boolean = false + + @Volatile internal var stopSendingAcked: Boolean = false internal data class ResetState( diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt index 646a4b58b..1c78e6eac 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt @@ -108,10 +108,10 @@ class SendBuffer { * by [takeChunk] (append), [markAcked] (remove / split), [markLost] * (remove and move to [retransmit]). * - * Range arithmetic is O(N) on the inFlight list; for the moq-rooms - * workload (a few thousand ranges max per connection) this is fine. - * If profiling later shows it on the hot path, swap in a TreeMap - * keyed by offset. + * Overlap lookup uses [firstOverlapIndex] (binary search, O(log N)); + * the early-exit forward walk visits only the k entries that actually + * overlap. A single ACK or loss notification therefore costs + * O(log N + k), with k typically 1-2 for a well-aligned ACK range. */ private val inFlight: ArrayDeque = ArrayDeque() @@ -394,16 +394,13 @@ class SendBuffer { } endIndex += 1 } - // Bulk-remove the contiguous overlap span [startIndex, endIndex). - // ArrayDeque doesn't expose subList.clear, so fall back to a - // tight removeAt loop walking backward — still O(k) per call - // but with a single shift of trailing entries instead of one - // shift per removed item. + // Remove the contiguous overlap span [startIndex, endIndex). + // ArrayDeque has no bulk-clear-range; each removeAt(i) shifts + // (size - i) elements, so total cost is O(k * (size - end + k)) + // worst case. Backward iteration shrinks the per-step shift — + // negligible for our workload (k is 1 in steady state, occasionally + // 2 across a partial-overlap split). if (endIndex > startIndex) { - // Walking backward minimises shift work: removeAt at the - // rightmost index of the span first leaves the trailing - // entries (after endIndex) where they are; only the leftward - // entries shift, and they shift by 1 each iteration. var i = endIndex - 1 while (i >= startIndex) { list.removeAt(i) diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt index 849c3bd00..d71ca978c 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ResetStopSendingEmitTest.kt @@ -183,6 +183,56 @@ class ResetStopSendingEmitTest { assertEquals(false, stream.stopSendingEmitPending) } + @Test + fun resetStream_secondCallIsNoOp_finalSizeFrozen() = + runBlocking { + // RFC 9000 §3.5: finalSize is fixed at first emission. + // A second resetStream() must not overwrite resetState + // — otherwise a retransmit after additional enqueue would + // replay with a larger finalSize, triggering FINAL_SIZE_ERROR. + val client = handshakedClient() + val stream = client.openUniStream() + stream.send.enqueue("first".encodeToByteArray()) // 5 bytes + stream.resetStream(errorCode = 1L) + + // App enqueues more bytes (writer's send loop is racing) and + // calls resetStream again with a different code. + stream.send.enqueue("more-bytes".encodeToByteArray()) // +10 bytes + stream.resetStream(errorCode = 99L) + + runCatching { drainOutbound(client, nowMillis = 1L) } + val tokenEntry = + client.application.sentPackets.entries + .firstOrNull { it.value.tokens.any { t -> t is RecoveryToken.ResetStream } } + assertNotNull(tokenEntry) + val token = + tokenEntry.value.tokens + .filterIsInstance() + .single() + assertEquals(1L, token.errorCode, "first errorCode wins") + assertEquals(5L, token.finalSize, "finalSize frozen at first call (RFC 9000 §3.5)") + } + + @Test + fun stopSending_secondCallIsNoOp() = + runBlocking { + val client = handshakedClient() + val stream = client.openBidiStream() + stream.stopSending(errorCode = 5L) + stream.stopSending(errorCode = 42L) + + runCatching { drainOutbound(client, nowMillis = 1L) } + val tokenEntry = + client.application.sentPackets.entries + .firstOrNull { it.value.tokens.any { t -> t is RecoveryToken.StopSending } } + assertNotNull(tokenEntry) + val token = + tokenEntry.value.tokens + .filterIsInstance() + .single() + assertEquals(5L, token.errorCode, "first errorCode wins") + } + @Test fun newConnectionId_retransmittedOnLoss() = runBlocking {