fix(quic): RESET_STREAM/STOP_SENDING first-call-wins + threading contract
Audit follow-up from the prior two commits.
1. resetStream() / stopSending() now no-op on the second call. RFC 9000
§3.5 pins finalSize at first emission; replaying retransmits with a
larger value (because the app enqueued more bytes between two
resetStream calls) would trigger FINAL_SIZE_ERROR on the peer. The
"idempotent: a second call overwrites with the newer error code"
claim was simply wrong. Two new tests lock the contract:
resetStream_secondCallIsNoOp_finalSizeFrozen and
stopSending_secondCallIsNoOp.
2. resetEmitPending / resetAcked / stopSendingEmitPending /
stopSendingAcked are now @Volatile. The public emit APIs are
callable from any coroutine while the writer / loss / ACK
dispatchers read the same fields under QuicConnection.lock; volatile
gives the cross-thread happens-before, and the first-call-wins gate
above eliminates the only multi-writer race (two app threads racing
the writer's clear-after-emit).
3. SendBuffer's class-level KDoc still claimed range arithmetic was
O(N) "swap to TreeMap if profiling flags it" — stale after the
binary-search refactor in 303caa8. Updated to reflect the actual
O(log N + k) cost.
4. The bulk-removal comment in removeOverlap overstated the win
("O(k) per call, single shift of trailing entries"). ArrayDeque
removeAt(i) shifts on every call, so the actual cost is
O(k * (size - end + k)). Toned down — it's still cheap because
k is 1-2 in steady state.
https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ
This commit is contained in:
@@ -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(
|
||||
|
||||
@@ -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<Range> = 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)
|
||||
|
||||
+50
@@ -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<RecoveryToken.ResetStream>()
|
||||
.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<RecoveryToken.StopSending>()
|
||||
.single()
|
||||
assertEquals(5L, token.errorCode, "first errorCode wins")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun newConnectionId_retransmittedOnLoss() =
|
||||
runBlocking {
|
||||
|
||||
Reference in New Issue
Block a user