nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(quic): extend RecoveryToken — Stream, Crypto, ResetStream, StopSending, NewConnectionId

Browse Source 
Token-shape commit (commit A of the deferred-follow-ups pass that
extends RFC 9002 retransmit beyond the receive-side flow-control
extensions originally shipped at commit 9e6fa3d).

New variants on RecoveryToken sealed class:
  - Stream(streamId, offset, length, fin): STREAM data range we
    sent. On loss the dispatcher will re-queue the range on the
    stream's SendBuffer (commits C, D wire that).
  - Crypto(level, offset, length): CRYPTO bytes at a specific
    encryption level. RFC 9000 §17.2.5 + §13.3: handshake bytes
    are reliable; lost CRYPTO retransmits at the same encryption
    level. Commit E wires the per-level dispatch.
  - ResetStream(streamId, errorCode, finalSize),
    StopSending(streamId, errorCode),
    NewConnectionId(seq, retirePriorTo, cid, statelessResetToken):
    reliable per RFC 9000 §13.3, scaffolding-only since :quic
    doesn't currently emit any of these. The pendingResetStream /
    pendingStopSending / pendingNewConnectionId maps on
    QuicConnection are populated by the dispatcher but not yet
    drained by any writer code path.

QuicConnection.onTokensLost extended with the new variants:
  - Stream/Crypto: route to the matching SendBuffer.markLost
    (currently a no-op — see SendBuffer.markLost kdoc; commit C
    replaces the stub with the retain-until-ACK rewrite).
  - ResetStream/StopSending/NewConnectionId: write to the
    pending* maps; future emit code drains them.

NewConnectionId data-class needs explicit equals/hashCode because
its ByteArray fields would otherwise compare by identity (Kotlin
data-class auto-equals limitation). Tested.

Tests added (4):
  - whenDispatch_isExhaustive extended with all 10 variants;
    catches at compile time if a new variant is added without
    updating the dispatcher
  - newConnectionId_arrayEqualityIsByContent
  - stream_equalityByValue
  - crypto_equalityByValue

SendBuffer gains placeholder markLost(offset, length, fin) and
markAcked(offset, length) methods — both no-ops today. Their
signatures are stable so the dispatcher wiring lands once now and
doesn't need re-touching when commit C makes them functional.

Full :quic test suite passes.

https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ
This commit is contained in:
Claude
2026-05-04 23:14:42 +00:00
parent c43c95184e
commit 7f6d9085a4
4 changed files with 279 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt
+64
View File
@@ -205,6 +205,31 @@ 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
* as [pendingResetStream].
*/
internal val pendingNewConnectionId:
MutableMap<Long, com.vitorpamplona.quic.connection.recovery.RecoveryToken.NewConnectionId> = HashMap()
/**
* RFC 9002 RTT estimator + loss-detection algorithm. Single
* shared instance per connection (RTT is per-path; we model a
@@ -787,6 +812,45 @@ class QuicConnection(
pendingMaxStreamData[token.streamId] = token.maxData
}
}
is com.vitorpamplona.quic.connection.recovery.RecoveryToken.Stream -> {
// Re-queue the lost byte range on the stream's send
// buffer. The next writer drain pulls from the
// retransmit queue first (FIFO ahead of new sends).
// See [com.vitorpamplona.quic.stream.SendBuffer.markLost].
val stream = streamByIdLocked(token.streamId) ?: continue
stream.send.markLost(
offset = token.offset,
length = token.length,
fin = token.fin,
)
}
is com.vitorpamplona.quic.connection.recovery.RecoveryToken.Crypto -> {
// Same shape, applied to the per-level CRYPTO buffer.
levelState(token.level).cryptoSend.markLost(
offset = token.offset,
length = token.length,
fin = false,
)
}
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
}
is com.vitorpamplona.quic.connection.recovery.RecoveryToken.StopSending -> {
pendingStopSending[token.streamId] = token
}
is com.vitorpamplona.quic.connection.recovery.RecoveryToken.NewConnectionId -> {
pendingNewConnectionId[token.sequenceNumber] = token
}
}
}
}
quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryToken.kt
+93
View File
@@ -95,4 +95,97 @@ sealed class RecoveryToken {
val streamId: Long,
val maxData: Long,
) : RecoveryToken()
/**
* STREAM data range we sent. On loss, the dispatcher informs the
* stream's [com.vitorpamplona.quic.stream.SendBuffer] that bytes
* `[offset, offset + length)` need retransmission — the buffer
* moves them from "sent" back to "needs retransmit", and the
* next [com.vitorpamplona.quic.connection.QuicConnectionWriter]
* drain re-emits them with the same offset (idempotent retransmit
* per RFC 9000 §13.3).
*
* [fin] tracks whether this STREAM frame carried the FIN bit;
* lost FIN frames must also be retransmitted (the FIN is part of
* the stream's reliable byte sequence).
*/
data class Stream(
val streamId: Long,
val offset: Long,
val length: Long,
val fin: Boolean,
) : RecoveryToken()
/**
* CRYPTO data range we sent at a specific encryption level.
* RFC 9000 §17.2.5: handshake bytes are reliable per RFC 9000
* §13.3; a lost CRYPTO frame must be retransmitted at the same
* encryption level it was originally sent at. The dispatcher
* consults [level] to route to the correct
* [com.vitorpamplona.quic.connection.LevelState.cryptoSend].
*/
data class Crypto(
val level: com.vitorpamplona.quic.connection.EncryptionLevel,
val offset: Long,
val length: Long,
) : RecoveryToken()
/**
* `RESET_STREAM` frame we sent (RFC 9000 §19.4). Carries the
* stream id, the application error code, and the final size.
* On loss, retransmit verbatim — RESET_STREAM is reliable per
* §13.3.
*
* `:quic` doesn't currently emit RESET_STREAM (no application
* code triggers stream reset), so this variant is scaffolding for
* future work. Listed in the enum so the dispatcher's `when`
* stays exhaustive at compile time.
*/
data class ResetStream(
val streamId: Long,
val errorCode: Long,
val finalSize: Long,
) : RecoveryToken()
/**
* `STOP_SENDING` frame we sent (RFC 9000 §19.5). Reliable per
* §13.3; retransmit verbatim on loss. Same scaffolding-only
* status as [ResetStream] — `:quic` doesn't emit it today.
*/
data class StopSending(
val streamId: Long,
val errorCode: Long,
) : RecoveryToken()
/**
* `NEW_CONNECTION_ID` frame we sent (RFC 9000 §19.15). Reliable
* per §13.3. Same scaffolding-only status — connection-ID
* rotation isn't wired today.
*/
data class NewConnectionId(
val sequenceNumber: Long,
val retirePriorTo: Long,
val connectionId: ByteArray,
val statelessResetToken: ByteArray,
) : RecoveryToken() {
// ByteArray fields require explicit equals/hashCode for data-class
// contract correctness — Kotlin's data-class auto-equals does
// identity compare on arrays.
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (other !is NewConnectionId) return false
return sequenceNumber == other.sequenceNumber &&
retirePriorTo == other.retirePriorTo &&
connectionId.contentEquals(other.connectionId) &&
statelessResetToken.contentEquals(other.statelessResetToken)
}
override fun hashCode(): Int {
var result = sequenceNumber.hashCode()
result = 31 * result + retirePriorTo.hashCode()
result = 31 * result + connectionId.contentHashCode()
result = 31 * result + statelessResetToken.contentHashCode()
return result
}
}
}
quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt
+44
View File
@@ -130,6 +130,50 @@ class SendBuffer {
Chunk(offset, data, fin)
}
/**
* Re-queue a previously-sent byte range for retransmission. Called
* by [com.vitorpamplona.quic.connection.QuicConnection.onTokensLost]
* when the [com.vitorpamplona.quic.connection.recovery.RecoveryToken.Stream]
* (or `Crypto`) token's carrying packet is declared lost.
*
* Currently a no-op: the buffer releases bytes on [takeChunk] (the
* "best-effort mode" documented at the top of this class), so by
* the time loss is detected the original bytes are gone and we
* have nothing to resend. The full retain-until-ACK rewrite lands
* in step C of the deferred-follow-ups pass started at commit
* `9e6fa3d` — once that's in, this method moves the [offset,
* offset + length) range from the "sent" set back to the
* "needs retransmit" queue, and the next [takeChunk] pulls from
* that queue first.
*
* The signature is stable now so the dispatcher doesn't need a
* second wiring pass when the implementation lands.
*/
fun markLost(
offset: Long,
length: Long,
fin: Boolean,
) {
// Step C placeholder. Parameters unused on purpose — the
// suppression makes the no-op intent explicit so a future
// reader doesn't think it's a bug.
@Suppress("UNUSED_PARAMETER")
val unusedForStepC = Triple(offset, length, fin)
}
/**
* Release ACK'd bytes from the retain-until-ACK retention buffer.
* Same step-C placeholder as [markLost]: today the buffer doesn't
* retain anything, so there's nothing to release.
*/
fun markAcked(
offset: Long,
length: Long,
) {
@Suppress("UNUSED_PARAMETER")
val unusedForStepC = Pair(offset, length)
}
data class Chunk(
val offset: Long,
val data: ByteArray,
quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryTokenTest.kt
+78
View File
@@ -98,6 +98,20 @@ class RecoveryTokenTest {
RecoveryToken.MaxStreamsBidi(150L),
RecoveryToken.MaxData(1_000_000L),
RecoveryToken.MaxStreamData(streamId = 0L, maxData = 1024L),
RecoveryToken.Stream(streamId = 4L, offset = 0L, length = 100L, fin = false),
RecoveryToken.Crypto(
level = com.vitorpamplona.quic.connection.EncryptionLevel.HANDSHAKE,
offset = 0L,
length = 64L,
),
RecoveryToken.ResetStream(streamId = 4L, errorCode = 0L, finalSize = 100L),
RecoveryToken.StopSending(streamId = 4L, errorCode = 0L),
RecoveryToken.NewConnectionId(
sequenceNumber = 1L,
retirePriorTo = 0L,
connectionId = byteArrayOf(1, 2, 3, 4),
statelessResetToken = ByteArray(16) { it.toByte() },
),
)
val labels =
tokens.map {
@@ -107,6 +121,11 @@ class RecoveryTokenTest {
is RecoveryToken.MaxStreamsBidi -> "msb:${it.maxStreams}"
is RecoveryToken.MaxData -> "md:${it.maxData}"
is RecoveryToken.MaxStreamData -> "msd:${it.streamId}:${it.maxData}"
is RecoveryToken.Stream -> "s:${it.streamId}:${it.offset}:${it.length}:${it.fin}"
is RecoveryToken.Crypto -> "c:${it.level}:${it.offset}:${it.length}"
is RecoveryToken.ResetStream -> "rs:${it.streamId}:${it.errorCode}:${it.finalSize}"
is RecoveryToken.StopSending -> "ss:${it.streamId}:${it.errorCode}"
is RecoveryToken.NewConnectionId -> "ncid:${it.sequenceNumber}"
}
}
assertEquals(
@@ -116,8 +135,67 @@ class RecoveryTokenTest {
"msb:150",
"md:1000000",
"msd:0:1024",
"s:4:0:100:false",
"c:HANDSHAKE:0:64",
"rs:4:0:100",
"ss:4:0",
"ncid:1",
),
labels,
)
}
@Test
fun newConnectionId_arrayEqualityIsByContent() {
// ByteArray fields must equal by content, not identity, so
// the sent-packet map's equality semantics work correctly.
val a =
RecoveryToken.NewConnectionId(
sequenceNumber = 1L,
retirePriorTo = 0L,
connectionId = byteArrayOf(1, 2, 3),
statelessResetToken = ByteArray(16) { 0 },
)
val b =
RecoveryToken.NewConnectionId(
sequenceNumber = 1L,
retirePriorTo = 0L,
connectionId = byteArrayOf(1, 2, 3),
statelessResetToken = ByteArray(16) { 0 },
)
assertEquals(a, b)
assertEquals(a.hashCode(), b.hashCode())
assertNotEquals(a, b.copy(connectionId = byteArrayOf(9, 9, 9)))
assertNotEquals(a, b.copy(statelessResetToken = ByteArray(16) { 7 }))
}
@Test
fun stream_equalityByValue() {
val t1 = RecoveryToken.Stream(streamId = 4L, offset = 0L, length = 100L, fin = false)
val t2 = RecoveryToken.Stream(streamId = 4L, offset = 0L, length = 100L, fin = false)
val tFin = RecoveryToken.Stream(streamId = 4L, offset = 0L, length = 100L, fin = true)
val tDifferentOffset = RecoveryToken.Stream(streamId = 4L, offset = 100L, length = 100L, fin = false)
assertEquals(t1, t2)
assertNotEquals(t1, tFin)
assertNotEquals(t1, tDifferentOffset)
}
@Test
fun crypto_equalityByValue() {
val a =
RecoveryToken.Crypto(
level = com.vitorpamplona.quic.connection.EncryptionLevel.INITIAL,
offset = 0L,
length = 64L,
)
val b =
RecoveryToken.Crypto(
level = com.vitorpamplona.quic.connection.EncryptionLevel.INITIAL,
offset = 0L,
length = 64L,
)
val differentLevel = a.copy(level = com.vitorpamplona.quic.connection.EncryptionLevel.HANDSHAKE)
assertEquals(a, b)
assertNotEquals(a, differentLevel)
}
}