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

fix(quic): correct AckTracker.purgeBelow via ACK-of-ACK dispatch

Browse Source 
Pre-fix, QuicConnectionParser purged the inbound AckTracker on every
inbound AckFrame using `frame.largestAcknowledged - frame.firstAckRange`
— but that value lives in OUR outbound PN space, while the tracker
holds inbound PNs we received from the peer. The two PN spaces are
unrelated; the bug mostly hid because they grow at similar rates,
but caused range-list bloat over long sessions where traffic is
asymmetric (e.g. listener receives ~50 audio frames/sec while sending
back ~1 ACK/sec).

The correct semantics: purge only when the peer has confirmed receipt
of OUR outbound ACK frame. Now driven by the ACK-of-ACK dispatch.

- RecoveryToken.Ack changed from data object to data class carrying
  (level, largestAcked) — the encryption level and the largest inbound
  PN our outbound ACK frame covered.
- QuicConnectionWriter populates these fields from the AckFrame at
  emit time.
- QuicConnection.onTokensAcked dispatches RecoveryToken.Ack to
  levelState(level).ackTracker.purgeBelow(largestAcked + 1).
- The wrong purge in QuicConnectionParser is removed (replaced with a
  comment pointing at the new dispatch path).

Listed in the audit-summary deferred-work as item 6
(`AckTracker.purgeBelow threshold semantics`).

New test: AckTrackerPurgeOnAckOfAckTest (4 cases — purge on
ack-of-ack, level routing, partial purge keeps higher PNs, out-of-order
ACKs are safe).

https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ
This commit is contained in:
Claude
2026-05-05 01:30:02 +00:00
parent 70af1953dc
commit e3a3ffd1d9
9 changed files with 249 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/AckTrackerPurgeOnAckOfAckTest.kt
+166
View File
@@ -0,0 +1,166 @@
/*
* 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 kotlinx.coroutines.runBlocking
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertTrue
/**
* RFC 9000 §13.2.1: an ACK frame is itself acknowledged via the
* carrying packet. Once the peer ACKs that packet, we know the peer
* received our ACK and we can drop the corresponding inbound PNs from
* our [com.vitorpamplona.quic.recovery.AckTracker].
*
* Pre-fix, `QuicConnectionParser` purged the tracker on every inbound
* ACK using `frame.largestAcknowledged - frame.firstAckRange` — but
* that value is in OUR outbound PN space, not the inbound PN space the
* tracker holds. The mistake mostly hid because both spaces grow at
* similar rates, but caused range-list bloat over long sessions where
* traffic is asymmetric (e.g. listener receives ~50 audio frames/sec
* while sending ~1 ACK/sec back).
*/
class AckTrackerPurgeOnAckOfAckTest {
private fun newConn(): QuicConnection =
QuicConnection(
serverName = "example.test",
config = QuicConnectionConfig(),
tlsCertificateValidator =
com.vitorpamplona.quic.tls
.PermissiveCertificateValidator(),
)
@Test
fun ackOfAck_purgesTrackerBelowLargestAcked() =
runBlocking {
val conn = newConn()
// Simulate having received 5 inbound packets from the peer.
for (pn in 0L..4L) {
conn.application.ackTracker.receivedPacket(pn, ackEliciting = true, receivedAtMillis = 1L)
}
assertEquals(4L, conn.application.ackTracker.largestReceived())
assertFalse(conn.application.ackTracker.isEmpty())
// Peer ACKs the packet that carried our outbound ACK
// covering up to PN 4.
conn.lock.lock()
try {
conn.onTokensAcked(
listOf(
RecoveryToken.Ack(
level = EncryptionLevel.APPLICATION,
largestAcked = 4L,
),
),
)
} finally {
conn.lock.unlock()
}
// Tracker is now empty: peer has confirmed receipt of our
// ACK that covered everything up to PN 4. Re-advertising
// those PNs in subsequent outbound ACKs is wasted bytes.
assertTrue(conn.application.ackTracker.isEmpty(), "tracker fully purged below largestAcked + 1")
}
@Test
fun ackOfAck_routesToCorrectLevel() =
runBlocking {
val conn = newConn()
// Independent state on Initial vs Application trackers.
for (pn in 0L..2L) {
conn.initial.ackTracker.receivedPacket(pn, ackEliciting = true, receivedAtMillis = 1L)
}
for (pn in 0L..4L) {
conn.application.ackTracker.receivedPacket(pn, ackEliciting = true, receivedAtMillis = 1L)
}
// Peer ACKs our Initial-level outbound ACK.
conn.lock.lock()
try {
conn.onTokensAcked(
listOf(
RecoveryToken.Ack(level = EncryptionLevel.INITIAL, largestAcked = 2L),
),
)
} finally {
conn.lock.unlock()
}
// Initial tracker drained; Application tracker untouched.
assertTrue(conn.initial.ackTracker.isEmpty())
assertEquals(4L, conn.application.ackTracker.largestReceived())
}
@Test
fun ackOfAck_partialPurge_keepsHigherPns() =
runBlocking {
val conn = newConn()
for (pn in 0L..9L) {
conn.application.ackTracker.receivedPacket(pn, ackEliciting = true, receivedAtMillis = 1L)
}
// Peer ACKs our outbound ACK that covered up to PN 4 only;
// the tracker's higher-PN ranges (5..9) must survive.
conn.lock.lock()
try {
conn.onTokensAcked(
listOf(
RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 4L),
),
)
} finally {
conn.lock.unlock()
}
assertFalse(conn.application.ackTracker.isEmpty())
assertEquals(9L, conn.application.ackTracker.largestReceived())
}
@Test
fun ackOfAck_outOfOrder_isSafe() =
runBlocking {
// Two outbound ACKs go out: ACK#A covers up-to PN 4, ACK#B
// covers up-to PN 9. Peer ACKs ACK#B first, then ACK#A.
// After ACK#B drains, tracker is empty. ACK#A's purge is
// a no-op — must not throw, must not re-resurrect anything.
val conn = newConn()
for (pn in 0L..9L) {
conn.application.ackTracker.receivedPacket(pn, ackEliciting = true, receivedAtMillis = 1L)
}
conn.lock.lock()
try {
conn.onTokensAcked(
listOf(
RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 9L),
),
)
assertTrue(conn.application.ackTracker.isEmpty())
conn.onTokensAcked(
listOf(
RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 4L),
),
)
assertTrue(conn.application.ackTracker.isEmpty())
} finally {
conn.lock.unlock()
}
}
}
quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/OnTokensLostTest.kt
+2 -2
View File
@@ -49,7 +49,7 @@ class OnTokensLostTest {
val conn = newConn()
conn.lock.lock()
try {
conn.onTokensLost(listOf(RecoveryToken.Ack))
conn.onTokensLost(listOf(RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 0L)))
} finally {
conn.lock.unlock()
}
@@ -161,7 +161,7 @@ class OnTokensLostTest {
conn.advertisedMaxData = 5_000_000L
conn.onTokensLost(
listOf(
RecoveryToken.Ack,
RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 0L),
RecoveryToken.MaxStreamsUni(maxStreams = 150L),
RecoveryToken.MaxStreamsBidi(maxStreams = 200L),
RecoveryToken.MaxData(maxData = 5_000_000L),
quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/RetransmitIntegrationTest.kt
+4 -1
View File
@@ -80,7 +80,10 @@ class RetransmitIntegrationTest {
sentAtMillis = 1L,
ackEliciting = true,
sizeBytes = 64,
tokens = listOf(RecoveryToken.Ack),
tokens =
listOf(
RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 0L),
),
)
// Drain the ACK'd packet from the map (simulating the
// parser path).
quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/RecoveryTokenTest.kt
+19 -10
View File
@@ -23,12 +23,11 @@ package com.vitorpamplona.quic.connection.recovery
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertNotEquals
import kotlin.test.assertTrue
/**
* Type-level invariants for [RecoveryToken]: equality + hashCode
* correctness across data-class variants, the singleton [Ack] object,
* and the typed-token shape required by the dispatcher in step 6 of
* correctness across data-class variants and the typed-token shape
* required by the dispatcher in step 6 of
* `quic/plans/2026-05-04-control-frame-retransmit.md`.
*
* No connection state is exercised here — these are unit tests on
@@ -36,11 +35,18 @@ import kotlin.test.assertTrue
*/
class RecoveryTokenTest {
@Test
fun ack_isSingleton() {
val a: RecoveryToken = RecoveryToken.Ack
val b: RecoveryToken = RecoveryToken.Ack
// `data object Ack` ⇒ same reference, same hash.
assertTrue(a === b)
fun ack_carriesLevelAndLargestAcked() {
val a: RecoveryToken =
RecoveryToken.Ack(
level = com.vitorpamplona.quic.connection.EncryptionLevel.APPLICATION,
largestAcked = 42L,
)
val b: RecoveryToken =
RecoveryToken.Ack(
level = com.vitorpamplona.quic.connection.EncryptionLevel.APPLICATION,
largestAcked = 42L,
)
// Data class equality, not identity.
assertEquals(a, b)
assertEquals(a.hashCode(), b.hashCode())
}
@@ -93,7 +99,10 @@ class RecoveryTokenTest {
// compiling — caught at build time, not at runtime.
val tokens: List<RecoveryToken> =
listOf(
RecoveryToken.Ack,
RecoveryToken.Ack(
level = com.vitorpamplona.quic.connection.EncryptionLevel.APPLICATION,
largestAcked = 0L,
),
RecoveryToken.MaxStreamsUni(150L),
RecoveryToken.MaxStreamsBidi(150L),
RecoveryToken.MaxData(1_000_000L),
@@ -116,7 +125,7 @@ class RecoveryTokenTest {
val labels =
tokens.map {
when (it) {
RecoveryToken.Ack -> "ack"
is RecoveryToken.Ack -> "ack"
is RecoveryToken.MaxStreamsUni -> "msu:${it.maxStreams}"
is RecoveryToken.MaxStreamsBidi -> "msb:${it.maxStreams}"
is RecoveryToken.MaxData -> "md:${it.maxData}"
quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/recovery/SentPacketTest.kt
+7 -3
View File
@@ -20,6 +20,7 @@
*/
package com.vitorpamplona.quic.connection.recovery
import com.vitorpamplona.quic.connection.EncryptionLevel
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertNotEquals
@@ -63,7 +64,7 @@ class SentPacketTest {
sentAtMillis = 0L,
ackEliciting = false,
sizeBytes = 16,
tokens = listOf(RecoveryToken.Ack),
tokens = listOf(RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 0L)),
)
val differentPn = a.copy(packetNumber = 8L)
assertNotEquals(a, differentPn)
@@ -95,10 +96,13 @@ class SentPacketTest {
sentAtMillis = 0L,
ackEliciting = false,
sizeBytes = 24,
tokens = listOf(RecoveryToken.Ack),
tokens = listOf(RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 0L)),
)
assertTrue(ackOnly.tokens.isNotEmpty())
assertEquals(RecoveryToken.Ack, ackOnly.tokens.single())
assertEquals(
RecoveryToken.Ack(level = EncryptionLevel.APPLICATION, largestAcked = 0L),
ackOnly.tokens.single(),
)
}
@Test