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 bc3dadef7..f0bdf77cc 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -205,6 +205,31 @@ 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]. + */ + internal val pendingNewConnectionId: + MutableMap = 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 + } } } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryToken.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryToken.kt index fae20eee6..65125fac3 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryToken.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryToken.kt @@ -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 + } + } } 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 58b42dde4..bc5192f20 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/SendBuffer.kt @@ -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, diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryTokenTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryTokenTest.kt index 135fa485b..5b952dd56 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryTokenTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryTokenTest.kt @@ -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) + } }