Merge pull request #2773 from vitorpamplona/claude/harden-quic-audio-rooms-kwqnS

feat(quic): retire fully-settled streams to keep tracker bounded under audio-room churn
This commit is contained in:
Vitor Pamplona
2026-05-07 20:19:13 -04:00
committed by GitHub
19 changed files with 3476 additions and 41 deletions
@@ -22,6 +22,7 @@ package com.vitorpamplona.amethyst
import android.app.Application
import com.vitorpamplona.amethyst.service.logging.Logging
import com.vitorpamplona.amethyst.service.nests.AppForegroundRecycleHook
import com.vitorpamplona.quartz.utils.Log
import com.vitorpamplona.quartz.utils.LogLevel
@@ -41,6 +42,17 @@ class Amethyst : Application() {
Log.d("AmethystApp") { "onCreate $this" }
instance = AppModules(this)
// After-background foreground recycle: when the app returns to
// the foreground after spending more than ~5 s in the
// background, publish a network-change event so every active
// NestViewModel recycles its underlying QUIC session. Covers
// the case where Android reclaims our UDP socket FD while
// backgrounded — the connectivity callback in
// `NestForegroundService` doesn't fire there because the
// network itself is still up. See `AppForegroundRecycleHook`'s
// kdoc for the threshold rationale.
registerActivityLifecycleCallbacks(AppForegroundRecycleHook())
if (isDebug) {
Logging.setup()
// Auto-enable the Nests session-trace recorder in debug
@@ -0,0 +1,173 @@
/*
* 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.amethyst.service.nests
import android.app.Activity
import android.app.Application
import android.os.Bundle
import com.vitorpamplona.amethyst.commons.viewmodels.NestNetworkChangeBus
import com.vitorpamplona.quartz.utils.Log
/**
* Pure-state foreground/background tracker, decoupled from
* `android.app.Activity` so the unit tests can drive it without
* Robolectric or Mockito.
*
* See [AppForegroundRecycleHook] for the production motivation /
* threshold-rationale kdoc — this class is just the testable core.
*
* Threading: all state-mutating methods are documented to run on the
* Android main thread (Application lifecycle callbacks fire there);
* tests call them serially, so no synchronisation is needed.
*/
class AppForegroundCounter(
private val backgroundThresholdMs: Long = AppForegroundRecycleHook.DEFAULT_BACKGROUND_THRESHOLD_MS,
private val publishEvent: () -> Unit = { NestNetworkChangeBus.publish() },
private val nowMillis: () -> Long = { System.currentTimeMillis() },
) {
private var startedActivities = 0
private var lastBackgroundedAtMillis: Long = -1L
/**
* Count of recycle events fired since construction. Diagnostic
* surface for tests; production code observes the side-effect via
* [NestNetworkChangeBus] instead.
*/
var recyclesFired: Int = 0
private set
/**
* Increment the started-activity counter and, if this is the
* 0 → 1 transition AND the app spent ≥ [backgroundThresholdMs]
* in the background, fire [publishEvent]. The first
* onActivityStarted after process start is a no-op (no prior
* background timestamp to compare against).
*/
fun onActivityStarted() {
val wasBackgrounded = startedActivities == 0
startedActivities++
if (!wasBackgrounded) return
val backgroundedAt = lastBackgroundedAtMillis
if (backgroundedAt < 0L) return
val backgroundedFor = nowMillis() - backgroundedAt
if (backgroundedFor < backgroundThresholdMs) {
Log.d("AppForegroundCounter") {
"skipping recycle on resume after only ${backgroundedFor}ms background " +
"(threshold=${backgroundThresholdMs}ms)"
}
return
}
Log.d("AppForegroundCounter") {
"publishing recycle event on resume after ${backgroundedFor}ms background"
}
recyclesFired++
publishEvent()
}
/**
* Decrement the counter; on N → 0 transition, record the
* background timestamp.
*/
fun onActivityStopped() {
startedActivities--
if (startedActivities <= 0) {
startedActivities = 0
lastBackgroundedAtMillis = nowMillis()
}
}
}
/**
* Application-wide observer that publishes a
* [NestNetworkChangeBus] event when the app returns to the foreground
* after spending more than [backgroundThresholdMs] in the background.
*
* Production motivation: Android may reclaim a backgrounded app's
* UDP-socket file descriptors as it ages out of the foreground app
* pool (the kernel's watcher trims after roughly 30 s of no foreground
* activity, with quite a bit of variance per OEM). When the user
* resumes the app, the QUIC connection sitting on the now-reclaimed
* socket has dead OS-level state, but the connection-level FSM
* doesn't know that yet — the next `socket.send` throws, and only
* then does the send-loop catch surface CLOSED.
*
* The downstream
* [com.vitorpamplona.amethyst.commons.viewmodels.NestViewModel]
* already observes [NestNetworkChangeBus] for the network-handover
* case (Wi-Fi ↔ cellular). We piggy-back on the same bus here: a
* "long enough background" event has the same shape from the QUIC
* driver's perspective as a network change — recycle the underlying
* session and let the [com.vitorpamplona.nestsclient.connectReconnectingNestsListener]
* / `connectReconnectingNestsSpeaker` orchestrators reconnect.
*
* Why a threshold instead of fire-on-every-resume:
* - A short background (notification pulldown, biometric auth, lock-
* screen glance) lasts < 1 s and the socket is still healthy. A
* forced recycle there is a wasted ~1 s re-handshake gap of audio
* silence — annoying to users for no benefit.
* - A long background (call from another app, screen off > 30 s,
* home-button-then-back) commonly leaves the socket dead. Better
* to eat one re-handshake gap than 30 s of silence while the QUIC
* PTO times out.
* 5 seconds is the sweet spot: well over typical UI transitions but
* well under any plausible socket-reclaim window.
*/
class AppForegroundRecycleHook(
backgroundThresholdMs: Long = DEFAULT_BACKGROUND_THRESHOLD_MS,
publishEvent: () -> Unit = { NestNetworkChangeBus.publish() },
nowMillis: () -> Long = { System.currentTimeMillis() },
) : Application.ActivityLifecycleCallbacks {
private val counter = AppForegroundCounter(backgroundThresholdMs, publishEvent, nowMillis)
override fun onActivityStarted(activity: Activity) {
counter.onActivityStarted()
}
override fun onActivityStopped(activity: Activity) {
counter.onActivityStopped()
}
override fun onActivityCreated(
activity: Activity,
savedInstanceState: Bundle?,
) = Unit
override fun onActivityResumed(activity: Activity) = Unit
override fun onActivityPaused(activity: Activity) = Unit
override fun onActivitySaveInstanceState(
activity: Activity,
outState: Bundle,
) = Unit
override fun onActivityDestroyed(activity: Activity) = Unit
companion object {
/**
* Default 5 000 ms — well above the longest plausible UI
* transition (notification pull, biometric prompt) and well
* below the 30 s timing the Android kernel uses to reclaim
* idle UDP sockets.
*/
const val DEFAULT_BACKGROUND_THRESHOLD_MS = 5_000L
}
}
@@ -0,0 +1,180 @@
/*
* 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.amethyst.service.nests
import org.junit.Assert.assertEquals
import org.junit.Test
/**
* Unit tests for [AppForegroundCounter] — the pure-state core of
* [AppForegroundRecycleHook]. Drives the lifecycle transitions
* directly with a controllable clock so the threshold logic doesn't
* need a real wall-clock wait.
*/
class AppForegroundRecycleHookTest {
@Test
fun firstForegroundAfterProcessStartDoesNotPublish() {
// The very first onActivityStarted has no prior background to
// recycle from — recycling here would fire a redundant
// re-handshake on every cold start, which is wasteful.
var fakeNow = 0L
var publishCount = 0
val counter =
AppForegroundCounter(
publishEvent = { publishCount++ },
nowMillis = { fakeNow },
)
fakeNow = 1_000L
counter.onActivityStarted()
assertEquals("first onActivityStarted must not publish — no prior background", 0, publishCount)
assertEquals(0, counter.recyclesFired)
}
@Test
fun shortBackgroundDoesNotTriggerRecycle() {
// 1 s background (notification pull, biometric prompt) is
// typical UI noise and the QUIC socket is still healthy.
var fakeNow = 0L
var publishCount = 0
val counter =
AppForegroundCounter(
backgroundThresholdMs = 5_000L,
publishEvent = { publishCount++ },
nowMillis = { fakeNow },
)
fakeNow = 1_000L
counter.onActivityStarted()
fakeNow = 2_000L
counter.onActivityStopped()
fakeNow = 3_000L // backgrounded for 1 s only
counter.onActivityStarted()
assertEquals(
"background < threshold must not publish — short transitions don't reclaim sockets",
0,
publishCount,
)
}
@Test
fun backgroundLongerThanThresholdTriggersRecycle() {
// 6 s background crosses the 5 s default threshold — Android
// may have reclaimed the socket FD by now, so recycle the
// QUIC session on resume.
var fakeNow = 0L
var publishCount = 0
val counter =
AppForegroundCounter(
backgroundThresholdMs = 5_000L,
publishEvent = { publishCount++ },
nowMillis = { fakeNow },
)
fakeNow = 1_000L
counter.onActivityStarted()
fakeNow = 2_000L
counter.onActivityStopped()
fakeNow = 8_000L // backgrounded for 6 s
counter.onActivityStarted()
assertEquals(
"background ≥ threshold must publish exactly once on resume",
1,
publishCount,
)
assertEquals(1, counter.recyclesFired)
}
@Test
fun multipleActivitiesTrackTransitionCorrectly() {
// Picture-in-picture mode is implemented as a second activity
// overlaid on the main activity. While both are started the
// app is in foreground; only when both stop does the app
// truly background.
var fakeNow = 0L
var publishCount = 0
val counter =
AppForegroundCounter(
backgroundThresholdMs = 5_000L,
publishEvent = { publishCount++ },
nowMillis = { fakeNow },
)
// First activity starts (cold start), no publish.
fakeNow = 1_000L
counter.onActivityStarted()
// Second activity (e.g. PIP / dialog) starts on top. Still
// foreground; no publish — `wasBackgrounded` was false.
fakeNow = 2_000L
counter.onActivityStarted()
// First activity stops (e.g. user backs out of main).
// counter = 1, still foreground.
fakeNow = 3_000L
counter.onActivityStopped()
assertEquals(
"intermediate stop with another activity still started must not background",
0,
publishCount,
)
// Second stops → app truly backgrounds.
fakeNow = 4_000L
counter.onActivityStopped()
// 6 s later, an activity restarts → recycle.
fakeNow = 10_000L
counter.onActivityStarted()
assertEquals(1, publishCount)
}
@Test
fun consecutiveLongBackgroundsEachPublishOnce() {
// Two separate back-and-forth cycles must each fire exactly
// one publish. A regression that misses to refresh the
// last-backgrounded timestamp on second-stop would either
// double-fire on the second resume or skip it.
var fakeNow = 0L
var publishCount = 0
val counter =
AppForegroundCounter(
backgroundThresholdMs = 5_000L,
publishEvent = { publishCount++ },
nowMillis = { fakeNow },
)
// Cycle 1: cold start → 6 s background → resume (publish #1)
fakeNow = 1_000L
counter.onActivityStarted()
fakeNow = 2_000L
counter.onActivityStopped()
fakeNow = 8_000L
counter.onActivityStarted()
assertEquals("first resume after long background must publish", 1, publishCount)
// Cycle 2: 8 s background again → resume (publish #2)
fakeNow = 10_000L
counter.onActivityStopped()
fakeNow = 18_000L
counter.onActivityStarted()
assertEquals("second resume after long background must also publish", 2, publishCount)
}
}
+11
View File
@@ -124,3 +124,14 @@ tasks.register<JavaExec>("interop") {
args(host, port)
systemProperty("interopTimeoutSec", timeoutSec)
}
// Long-form audio-rooms soak test. Disabled by default — `./gradlew test`
// must stay fast for CI. Opt in with `-PquicSoakSeconds=N` (e.g. 1800 for
// the 30-minute run from the soak prompt). Without the property the test
// class checks for null and skips via `Assume.assumeTrue`.
tasks.withType<Test>().configureEach {
val soakSeconds = (project.findProperty("quicSoakSeconds") as? String)
if (soakSeconds != null) {
systemProperty("quicSoakSeconds", soakSeconds)
}
}
@@ -301,15 +301,82 @@ class QuicConnection(
/**
* Round-4 perf #10: parallel insertion-ordered list of streams so the
* writer's round-robin scan can index by position without
* `streams.entries.toList()` allocating per drain. Streams are only ever
* added (no removal in the current model), so the two stay in sync as
* long as `getOrCreatePeerStreamLocked` and `openBidi/UniStream` append
* to both.
* `streams.entries.toList()` allocating per drain. Mutated in lockstep
* with [streams] under [streamsLock]:
* - [openBidiStreamLocked] / [openUniStreamLocked] /
* [getOrCreatePeerStreamLocked] append.
* - [retireFullyDoneStreamsLocked] removes entries whose stream has
* flipped [QuicStream.isFullyRetired] = true. The retire pass
* folds the receive-side high-water mark into
* [retiredStreamsRecvBytes] so the writer's MAX_DATA accounting
* in `appendFlowControlUpdates` doesn't regress when a retired
* stream's `receive.contiguousEnd()` drops out of the iteration.
*
* Without retirement, the moq-lite audio-rooms path leaks one
* QuicStream per Opus frame for the lifetime of the session — the
* stream-cliff investigation in
* `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`
* pegs steady-state churn at ~50 streams/sec, so a 3-hour room
* accumulates ~540 000 entries before retirement was wired.
*/
private val streamsList = mutableListOf<QuicStream>()
private var nextLocalBidiIndex: Long = 0L
private var nextLocalUniIndex: Long = 0L
/**
* Cumulative `receive.contiguousEnd()` from streams that have been
* removed by [retireFullyDoneStreamsLocked]. Folded into the writer's
* `totalRecvAdvanced` accumulator so MAX_DATA continues to advertise
* the *connection-lifetime* receive high-water mark even after the
* contributing streams have been dropped from [streamsList]. Without
* this, retiring N streams that each delivered K bytes would silently
* regress the advertised connection-level credit by N*K, eventually
* starving the peer once `advertisedMaxData` was tripped past.
*
* Caller of any read/write must hold [streamsLock].
*/
internal var retiredStreamsRecvBytes: Long = 0L
private set
/**
* Cumulative count of streams ever removed by
* [retireFullyDoneStreamsLocked]. Diagnostic-only; tests that drive
* stream churn use this to assert retirement actually fired (rather
* than relying on `streamsList.size` alone, which can shrink because
* a soak loop happened to drain the same workload it just enqueued).
*
* Caller of any read/write must hold [streamsLock].
*/
internal var retiredStreamsCount: Long = 0L
private set
/**
* FIFO ring of recently-retired stream IDs. The parser consults
* this in [getOrCreatePeerStreamLocked] to drop duplicate STREAM
* frames the peer might retransmit on a stream we've already torn
* down — without this guard, a retransmit (which can happen if our
* ACK of the FIN frame was lost and the peer's loss detector
* re-fired) would create a fresh QuicStream object, deliver
* duplicate bytes to the application, and bump
* [peerInitiatedUniCount] a second time.
*
* Bounded at [RETIRED_STREAM_ID_RING_SIZE] entries. Eviction is
* FIFO so a long-running session never grows this set unbounded —
* at moq-lite churn rates of ~50 streams/sec, the ring covers the
* last ~80 seconds of retired IDs, far longer than the peer's
* loss-detection retransmit window (a small multiple of RTT).
*
* Out-of-bounds duplicate retransmits (extremely rare — would need
* the peer's ACK to be lost AND our retransmit not arriving for
* many seconds) fall through to the existing
* "create-and-immediately-retire" path, which is the previous
* pre-guard behavior and merely costs a re-iteration.
*
* Caller of any read/write must hold [streamsLock].
*/
private val retiredStreamIdsOrder = ArrayDeque<Long>()
private val retiredStreamIdSet = HashSet<Long>()
/**
* Peer-advertised concurrent bidirectional stream cap. Initialised from
* [TransportParameters.initialMaxStreamsBidi] when peer params arrive,
@@ -413,6 +480,31 @@ class QuicConnection(
internal val pendingNewConnectionId:
MutableMap<Long, com.vitorpamplona.quic.connection.recovery.RecoveryToken.NewConnectionId> = HashMap()
/**
* RFC 9000 §8.2.2: queue of inbound PATH_CHALLENGE payloads we
* still owe a PATH_RESPONSE for. Each entry is the EXACT 8 bytes
* the peer challenged with — the response MUST echo them
* unchanged. The writer drains this queue on the next
* application-level packet build, emitting one PATH_RESPONSE
* per entry until empty.
*
* Bounded at [MAX_PENDING_PATH_RESPONSES] to defend against an
* attacker spamming PATH_CHALLENGE frames to exhaust memory.
* Excess challenges are dropped; the protocol allows it (a peer
* that doesn't get a response just retries).
*
* RFC 9000 §8.2.1 nuance: the response MUST go out on the
* incoming-packet's path. We model a single path today (one
* UDP socket per connection — see [com.vitorpamplona.quic.transport.UdpSocket]'s
* "no migration" kdoc), so "path the challenge arrived on" is
* trivially "the only path." When client-initiated migration
* lands, this queue grows to remember which path each entry
* belongs to.
*
* Caller must hold [streamsLock] for any read/write.
*/
internal val pendingPathChallengePayloads: ArrayDeque<ByteArray> = ArrayDeque()
/**
* RFC 9002 RTT estimator + loss-detection algorithm. Single
* shared instance per connection (RTT is per-path; we model a
@@ -1281,6 +1373,14 @@ class QuicConnection(
val wasClosed = status == Status.CLOSED
if (status != Status.CLOSED) status = Status.CLOSED
if (!wasClosed) {
// First-call wins for [closeReason] so the highest-quality
// diagnostic is preserved when several teardown paths race
// (e.g. read loop's `socket.receive() == null` finally fires
// a moment before the send loop's `socket.send` throw catch
// block does). Without this, downstream observers like
// `ReconnectingNestsListener.terminalAwait` see a closed
// connection but no human-readable cause for the failure.
closeReason = reason
// "remote" covers both peer-initiated CONNECTION_CLOSE and
// local invariant violations (CID mismatch, frame decode
// failure) that the parser surfaces as markClosedExternally.
@@ -1685,11 +1785,135 @@ class QuicConnection(
/**
* Insertion-ordered list view used by the writer's round-robin scan.
* Stays in sync with [streams] because the only mutation paths
* (openBidi/UniStream, getOrCreatePeerStreamLocked) append to both. No
* remove path exists today; if/when one is added it MUST update both.
* (openBidi/UniStream, getOrCreatePeerStreamLocked,
* retireFullyDoneStreamsLocked) update both.
*/
internal fun streamsListLocked(): List<QuicStream> = streamsList
/**
* Walk [streamsList] once and remove every stream whose
* [QuicStream.isFullyRetired] flag is true. Drops the entry from both
* [streamsList] and [streams]. Receive-side high-water marks fold
* into [retiredStreamsRecvBytes] so the writer's connection-level
* MAX_DATA accounting continues to see the lifetime total.
*
* Returns the number of streams removed in this pass.
*
* Caller MUST hold [streamsLock]. The writer drains under
* `streamsLock`, so calling this from the top of `buildApplicationPacket`
* is the natural place — it runs once per send pass, sees the latest
* FIN/RESET ACK state from the parser's just-finished processing, and
* the iteration order matches the writer's per-pass round-robin
* (so the rotation cursor doesn't accidentally point at a hole).
*
* Why retirement is safe even though the QUIC spec requires the peer
* to deliver retransmits if its ACK never reached us:
* - SEND side waits for `finAcked` / `resetAcked`, i.e. the peer has
* already confirmed receipt of our FIN. After that point the peer
* will not re-send anything on the stream.
* - RECEIVE side waits for the parser to have pushed every byte to
* the application's incoming Channel ([ReceiveBuffer.isFullyRead]).
* Retired stream IDs are recorded in [retiredStreamIdSet] so a
* duplicate STREAM frame the peer retransmits (because its own
* loss-detection fired before our ACK arrived) is dropped at
* [getOrCreatePeerStreamLocked] rather than minting a phantom
* stream that delivers duplicate bytes to the application.
*/
internal fun retireFullyDoneStreamsLocked(): Int {
if (streamsList.isEmpty()) return 0
var removed = 0
val it = streamsList.iterator()
while (it.hasNext()) {
val stream = it.next()
if (!stream.isFullyRetired) continue
// Fold the per-stream receive high-water into the cumulative
// counter BEFORE we drop it — once we lose the reference the
// writer can no longer reconstruct the contribution.
retiredStreamsRecvBytes += stream.receive.contiguousEnd()
// Defence-in-depth: ensure the application-side incoming
// channel is closed even if the parser somehow missed the
// closeIncoming call (e.g. a stream that finished via
// resetAcked never having received any STREAM frame at all).
// closeIncoming is idempotent.
stream.closeIncoming()
// Record the id so a peer retransmit on this stream gets
// dropped instead of recreating a phantom stream. FIFO
// ring with bounded size — ancient retired IDs eventually
// age out, but the eviction window is far larger than the
// peer's plausible retransmit horizon.
recordRetiredStreamIdLocked(stream.streamId)
streams.remove(stream.streamId)
it.remove()
removed++
}
if (removed > 0) {
retiredStreamsCount += removed.toLong()
// The writer's round-robin cursor is a position in
// [streamsList], so a removal that crossed the cursor would
// make the next drain pass skip a tier. Reset to 0 — the
// round-robin is just a fairness hint, not a semantic
// requirement.
streamRoundRobinStart = 0
}
return removed
}
/**
* Add [streamId] to the retired-IDs ring. FIFO eviction once the
* ring is at [RETIRED_STREAM_ID_RING_SIZE] entries — the oldest
* entry is removed from both the ordered deque and the lookup
* set in lockstep. Idempotent on duplicate adds (the
* [retireFullyDoneStreamsLocked] caller already drops the stream
* from `streams` before this fires, so a duplicate add can only
* happen if a caller manually re-records — defensive `add` skip
* keeps the ring entries unique without the cost of a fresh
* dedup pass).
*
* Caller must hold [streamsLock].
*/
private fun recordRetiredStreamIdLocked(streamId: Long) {
if (retiredStreamIdSet.add(streamId)) {
retiredStreamIdsOrder.addLast(streamId)
while (retiredStreamIdsOrder.size > RETIRED_STREAM_ID_RING_SIZE) {
val evicted = retiredStreamIdsOrder.removeFirst()
retiredStreamIdSet.remove(evicted)
}
}
}
/**
* Queue a PATH_RESPONSE for the given [challengeData]. Called by
* the parser when a PATH_CHALLENGE arrives. Idempotent on
* duplicate challenges (peer retransmit) — the writer drains
* each entry exactly once, so an over-eager peer that sends
* many PATH_CHALLENGEs gets many PATH_RESPONSEs (RFC 9000 §8.2
* permits this; the spec just requires at-least-one).
*
* The queue is bounded at [MAX_PENDING_PATH_RESPONSES]; excess
* entries are silently dropped (the protocol allows the peer
* to time out and retry).
*
* Caller must hold [streamsLock].
*/
internal fun queuePathResponseLocked(challengeData: ByteArray) {
if (pendingPathChallengePayloads.size >= MAX_PENDING_PATH_RESPONSES) return
// The parser produces a fresh ByteArray per PATH_CHALLENGE
// (see [com.vitorpamplona.quic.Buffer.QuicReader.readBytes],
// which `copyOfRange`s a slice off the inbound payload), so
// we can keep the reference directly without a defensive
// copy.
pendingPathChallengePayloads.addLast(challengeData)
}
/**
* True if [streamId] has been retired *and* is still inside the
* ring's eviction window. Used by [QuicConnectionParser] to drop
* STREAM frames the peer retransmits on already-retired streams.
*
* Caller must hold [streamsLock].
*/
internal fun isStreamIdRetiredLocked(streamId: Long): Boolean = streamId in retiredStreamIdSet
/** Caller must hold [lock]. Pending datagram queue for the driver's send loop. */
internal fun pendingDatagramsLocked(): ArrayDeque<ByteArray> = pendingDatagrams
@@ -1879,6 +2103,37 @@ class QuicConnection(
* audio/video, fresh frames matter more than stale ones.
*/
const val MAX_INCOMING_DATAGRAM_QUEUE: Int = 256
/**
* Capacity of the retired-stream-IDs ring used by
* [recordRetiredStreamIdLocked] / [isStreamIdRetiredLocked].
* Sized so the ring covers ≥ 80 seconds of retirement at
* moq-lite's ~50 streams/sec churn rate, far longer than any
* plausible peer retransmit window (a small multiple of RTT
* — at the absolute worst tens of seconds on lossy mobile
* networks). Older retired IDs eviction-fall-through to the
* existing create-and-immediately-retire path, which is
* functionally correct, just with one extra round of work
* per duplicate.
*
* Memory: 4 096 × (8 bytes Long key + ~32 bytes HashSet
* overhead + 8 bytes ArrayDeque slot) ≈ 200 KB. Trivial vs
* the per-stream object size we're saving by retiring.
*/
const val RETIRED_STREAM_ID_RING_SIZE: Int = 4_096
/**
* Bound on the [pendingPathChallengePayloads] queue. RFC 9000 §8.2
* doesn't cap PATH_CHALLENGE rate, so a malicious peer could
* spam them to exhaust our memory. 64 entries × 8 bytes = 512 B
* worst case — trivial to absorb but tight enough that an
* attacker can't pin 100 MB by flooding.
*
* Excess challenges are dropped; the spec allows it (a peer
* that doesn't see a response will retransmit on the next
* PTO if path validation matters to them).
*/
const val MAX_PENDING_PATH_RESPONSES: Int = 64
}
}
@@ -64,6 +64,30 @@ class QuicConnectionDriver(
private var readJob: Job? = null
private var sendJob: Job? = null
/**
* Test-only handle on the driver's [SupervisorJob]. Used by the
* session-lifecycle leak test in
* [com.vitorpamplona.quic.connection.QuicConnectionDriverLifecycleTest]
* to assert that a closed driver reports `isCompleted = true` once
* its read/send loops have unwound — the inverse of "the driver
* leaked a coroutine past close()".
*
* Production code MUST NOT touch this — the driver lifecycle is
* managed end-to-end by [start] / [close].
*/
internal val driverJob: Job get() = job
/**
* Test-only handle on the in-flight teardown coroutine. Returns null
* before [close] has been called; once close runs, the returned Job
* lets the test `join()` until teardown is complete (cancel + socket
* close + read/send join). Pre-existing close() returned immediately
* and provided no synchronous "teardown is done" signal — tests had
* to poll `connection.status == CLOSED` and trust that the rest of
* the cleanup eventually settled.
*/
internal val closeTeardownJob: Job? get() = closeJob
/**
* Round-5 concurrency #5: close() guard. A second concurrent invocation
* (e.g. session close + read-loop death close racing) used to launch a
@@ -127,6 +151,34 @@ class QuicConnectionDriver(
// the first RTT sample we fall back to a 1 s conservative
// floor (the same prior-shipping behavior, kept for
// handshake-timeout safety on lossy paths).
//
// Mirror the read loop's symmetry: any uncaught throw inside
// the loop (most common: `socket.send` raising once the OS
// tears down the UDP socket — typical on Android when the
// app backgrounds and the kernel reclaims the FD ~30 s
// later) MUST flip the connection to CLOSED so the
// higher-level reconnect orchestration in
// [com.vitorpamplona.nestsclient.connectReconnectingNestsListener]
// observes a Failed terminal state and fires a fresh
// handshake. Pre-fix, the throw escaped silently into the
// SupervisorJob, the read loop kept blocking on
// `socket.receive()` (which doesn't throw — it returns null
// on close, but the OS may not surface that for many
// seconds), and the connection sat in HANDSHAKING / CONNECTED
// long after the socket was dead — invisibly wedged.
try {
sendLoopBody()
} catch (ce: kotlinx.coroutines.CancellationException) {
// Cooperative cancel from close() / scope.cancel(). Don't
// mark closed here — close() is already driving the
// teardown and would race with our markClosedExternally.
throw ce
} catch (t: Throwable) {
connection.markClosedExternally("send loop exited: ${t::class.simpleName}: ${t.message}")
}
}
private suspend fun sendLoopBody() {
while (connection.status != QuicConnection.Status.CLOSED) {
// Phase 1 of the lock-split refactor: the writer holds
// streamsLock for the build, releases it for the actual
@@ -32,6 +32,8 @@ import com.vitorpamplona.quic.frame.MaxStreamDataFrame
import com.vitorpamplona.quic.frame.MaxStreamsFrame
import com.vitorpamplona.quic.frame.NewConnectionIdFrame
import com.vitorpamplona.quic.frame.NewTokenFrame
import com.vitorpamplona.quic.frame.PathChallengeFrame
import com.vitorpamplona.quic.frame.PathResponseFrame
import com.vitorpamplona.quic.frame.PingFrame
import com.vitorpamplona.quic.frame.ResetStreamFrame
import com.vitorpamplona.quic.frame.StopSendingFrame
@@ -301,26 +303,75 @@ private fun feedShortHeaderPacket(
return
}
val keysToUse: PacketProtection
// Build an ordered list of (keys, rotateOnSuccess) candidates and
// try AEAD against each in order. The list shape depends on whether
// the peer's KEY_PHASE bit matches our current phase:
//
// match → [current keys] (no rotation possible)
// mismatch → [previous keys (if any), next-phase keys (if derivable)]
//
// The mismatch path tries `previousReceiveProtection` FIRST because
// a reordered packet from before the last rotation is by far the
// common case (the reorder window is short, on the order of a few
// RTTs). If those keys decrypt the packet, we use them and don't
// commit. If they fail (genuine consecutive rotation — peer
// rotated AGAIN, KEY_PHASE wraps back to its prior value), we
// fall through to deriving next-phase keys against the
// already-rolled-forward `appReceiveSecret`. Two AEAD attempts on
// a mismatched-phase packet are OK; KEY_PHASE mismatch is rare
// (at most once per a-billion-packets in normal usage), and the
// failed first attempt is cheap (single AEAD seal-verify on a
// small payload).
//
// Pre-fix the parser routed every mismatch packet UNCONDITIONALLY
// through previousReceiveProtection if non-null, with no
// fallback. After two consecutive rotations the prior-keys path
// would always be the wrong keys, AEAD-fail, and the connection
// would silently wedge — `KeyUpdatePeerInitiatedTest`'s
// `twoConsecutiveRotationsCommitCorrectly` was disabled with a
// KNOWN-LIMITATION comment for this reason.
val parsed: ShortHeaderPacket.ParseResult?
val rotateOnSuccess: PacketProtection?
when {
peek.keyPhase == conn.currentReceiveKeyPhase -> {
keysToUse = live
if (peek.keyPhase == conn.currentReceiveKeyPhase) {
parsed =
ShortHeaderPacket.parseAndDecrypt(
bytes = datagram,
offset = offset,
dcidLen = conn.sourceConnectionId.length,
aead = live.aead,
key = live.key,
iv = live.iv,
hp = live.hp,
hpKey = live.hpKey,
largestReceivedInSpace = state.pnSpace.largestReceived,
)
rotateOnSuccess = null
} else {
// Try previous-phase first (reorder path). null result here
// means the packet wasn't from before the last rotation —
// fall through to next-phase derivation.
val priorTry =
conn.previousReceiveProtection?.let { prev ->
ShortHeaderPacket.parseAndDecrypt(
bytes = datagram,
offset = offset,
dcidLen = conn.sourceConnectionId.length,
aead = prev.aead,
key = prev.key,
iv = prev.iv,
hp = prev.hp,
hpKey = prev.hpKey,
largestReceivedInSpace = state.pnSpace.largestReceived,
)
}
if (priorTry != null) {
parsed = priorTry
rotateOnSuccess = null
}
// Reordered packet from before our last rotation — try the
// retained previous keys. Reordering window is small but real
// for paths with non-trivial RTT.
conn.previousReceiveProtection != null -> {
keysToUse = conn.previousReceiveProtection!!
rotateOnSuccess = null
}
// Peer just rotated. Derive next-phase keys and prepare to commit
// if AEAD succeeds. Failure path is the same as a corrupted /
// unauthenticated packet — silent drop.
else -> {
} else {
// Peer rotated (possibly for a 2nd time). Derive next-phase
// keys against the rolled-forward `appReceiveSecret` and
// try AEAD. On success we commit the rotation; on failure
// it's a bona-fide corrupt / unauthenticated packet.
val nextPhase = conn.deriveNextPhaseReceiveKeys()
if (nextPhase == null) {
conn.qlogObserver.onPacketDropped(
@@ -329,23 +380,21 @@ private fun feedShortHeaderPacket(
)
return
}
keysToUse = nextPhase
parsed =
ShortHeaderPacket.parseAndDecrypt(
bytes = datagram,
offset = offset,
dcidLen = conn.sourceConnectionId.length,
aead = nextPhase.aead,
key = nextPhase.key,
iv = nextPhase.iv,
hp = nextPhase.hp,
hpKey = nextPhase.hpKey,
largestReceivedInSpace = state.pnSpace.largestReceived,
)
rotateOnSuccess = nextPhase
}
}
val parsed =
ShortHeaderPacket.parseAndDecrypt(
bytes = datagram,
offset = offset,
dcidLen = conn.sourceConnectionId.length,
aead = keysToUse.aead,
key = keysToUse.key,
iv = keysToUse.iv,
hp = keysToUse.hp,
hpKey = keysToUse.hpKey,
largestReceivedInSpace = state.pnSpace.largestReceived,
)
if (parsed == null) {
conn.qlogObserver.onPacketDropped(
"AEAD auth failed or header parse failed at level APPLICATION",
@@ -513,6 +562,24 @@ private fun dispatchFrames(
)
return
}
// Phantom-stream guard: drop STREAM frames the peer
// retransmitted on a stream we've already retired. Without
// this, the next line would mint a fresh QuicStream object
// and re-deliver duplicate bytes to the application
// (worse, on a SERVER_UNI stream it would also bump
// peerInitiatedUniCount and trigger a spurious
// MAX_STREAMS_UNI emission). The frame is still
// ack-eliciting — we set ackEliciting above — so our
// ACK goes out and the peer's loss-detector backs off.
// Stays inside the StreamFrame handler so other frame
// types (RESET_STREAM, MAX_STREAM_DATA, STOP_SENDING)
// on retired ids gracefully fall through their existing
// streamByIdLocked == null branches as no-ops.
if (conn.isStreamIdRetiredLocked(frame.streamId) &&
conn.streamByIdLocked(frame.streamId) == null
) {
continue
}
val stream = conn.getOrCreatePeerStreamLocked(frame.streamId)
// RFC 9000 §4.1: peer MUST NOT send beyond the limit we advertised.
// The connection-level kill protects against unbounded memory
@@ -645,6 +712,39 @@ private fun dispatchFrames(
ackEliciting = true
}
is PathChallengeFrame -> {
// RFC 9000 §8.2.2 — peer is validating that a path is
// alive. We MUST echo the SAME 8-byte payload in a
// PATH_RESPONSE on the path the challenge arrived on.
// The writer drains [pendingPathChallengePayloads] on the next
// application-level packet build.
//
// Common practical trigger: server-side path
// validation after our connection-id rotation, OR
// post-NAT-rebind probing. Without responding the
// server may declare the path dead within a few RTTs
// and tear the connection down — visible to users as
// a sudden audio cut on a phone that briefly
// switched cells.
//
// RFC 9000 §13.2.1: PATH_CHALLENGE is ack-eliciting.
// The PATH_RESPONSE we queue here is itself
// ack-eliciting; the regular ACK path covers both.
ackEliciting = true
conn.queuePathResponseLocked(frame.data)
}
is PathResponseFrame -> {
// RFC 9000 §13.2.1: PATH_RESPONSE is ack-eliciting.
// We don't yet issue PATH_CHALLENGE ourselves (that's
// the client-initiated migration path, out of scope
// for the first-pass landing here), so any PATH_RESPONSE
// we receive is necessarily for a challenge we never
// sent — drop it after marking ack-eliciting so the
// outbound ACK still goes out.
ackEliciting = true
}
is ConnectionCloseFrame -> {
// Audit-4 #13: any frames following CONNECTION_CLOSE in the
// same payload MUST NOT be dispatched — they could create
@@ -32,6 +32,7 @@ import com.vitorpamplona.quic.frame.MaxDataFrame
import com.vitorpamplona.quic.frame.MaxStreamDataFrame
import com.vitorpamplona.quic.frame.MaxStreamsFrame
import com.vitorpamplona.quic.frame.NewConnectionIdFrame
import com.vitorpamplona.quic.frame.PathResponseFrame
import com.vitorpamplona.quic.frame.PingFrame
import com.vitorpamplona.quic.frame.ResetStreamFrame
import com.vitorpamplona.quic.frame.StopSendingFrame
@@ -575,6 +576,18 @@ private fun buildApplicationPacket(
// ServerHello has been processed.
val use1Rtt = state.sendProtection != null
val proto = state.sendProtection ?: conn.zeroRttSendProtection ?: return null
// Drop fully-settled streams BEFORE iterating so we don't waste a
// round-robin slot on a stream with nothing to send and no receive
// bookkeeping pending. Run this once per drain (at the top of the
// application-level path, the only level where streams exist) under
// the same `streamsLock` the rest of the drain holds — see
// [QuicConnection.retireFullyDoneStreamsLocked] for the safety
// argument. The cumulative receive bytes from removed streams fold
// into `appendFlowControlUpdates` below so MAX_DATA stays correct.
// Safe to call on the 0-RTT path too: no stream can be fully
// settled mid-handshake (peer hasn't ACK'd anything yet), so the
// walk is a no-op.
conn.retireFullyDoneStreamsLocked()
val frames = mutableListOf<Frame>()
// Tokens collected in lock-step with [frames]: each retransmittable
// frame contributes a [RecoveryToken] so the [SentPacket] recorded
@@ -937,6 +950,18 @@ private fun appendFlowControlUpdates(
}
}
// PATH_RESPONSE — drain every pending challenge response in one
// pass. RFC 9000 §13.3 is silent on retransmission for
// PATH_RESPONSE: it's NOT in the ack-eliciting-and-retransmittable
// class, so we don't track tokens for these. If the response
// packet is lost, the peer's next PATH_CHALLENGE retry queues a
// fresh entry here and we respond again. The peer is responsible
// for retrying its challenge until it sees a matching response.
while (conn.pendingPathChallengePayloads.isNotEmpty()) {
val data = conn.pendingPathChallengePayloads.removeFirst()
frames += PathResponseFrame(data)
}
// NEW_CONNECTION_ID retransmits. No application path emits these
// initially today (connection-ID rotation isn't wired); the map
// is populated only by the loss dispatcher, so this branch only
@@ -957,7 +982,15 @@ private fun appendFlowControlUpdates(
}
val cfg = conn.config
var totalRecvAdvanced = 0L
// Seed with the cumulative receive high-water mark from streams that
// [QuicConnection.retireFullyDoneStreamsLocked] has already dropped
// — the writer's connection-level MAX_DATA threshold uses the
// *lifetime* total, so retired streams must keep contributing even
// after their per-object `receive.contiguousEnd()` is no longer
// reachable. Without this seed, retiring K bytes of streams would
// silently regress the advertised credit by K and eventually starve
// the peer once the running total fell behind `advertisedMaxData`.
var totalRecvAdvanced = conn.retiredStreamsRecvBytes
// Round-4 perf #9 + round-5 #9: walk the streams via the index-friendly
// list view (no `entries.toList()` allocation), and only do per-stream
// window/threshold work for streams flagged by the parser since the last
@@ -313,6 +313,54 @@ class NewConnectionIdFrame(
}
}
/**
* RFC 9000 §19.17 — PATH_CHALLENGE frame, used for path validation
* (§8.2). The 8-byte [data] payload is opaque random bytes the
* sender uses to bind a PATH_RESPONSE back to a specific
* challenge. The receiver MUST echo the SAME 8 bytes in a
* [PathResponseFrame] on the path the challenge arrived on.
*
* Path validation lets either endpoint confirm the peer can still
* receive on a 4-tuple: the most common practical use is the
* server probing the client after a NAT rebind / connection
* migration. Without responding, the server may declare the path
* dead and tear the connection down — visible to users as a
* sudden audio cut on a phone that briefly switched cells.
*/
class PathChallengeFrame(
val data: ByteArray,
) : Frame() {
init {
require(data.size == 8) { "PATH_CHALLENGE data must be exactly 8 bytes per RFC 9000 §19.17" }
}
override fun encode(out: QuicWriter) {
out.writeByte(FrameType.PATH_CHALLENGE.toInt())
out.writeBytes(data)
}
}
/**
* RFC 9000 §19.18 — PATH_RESPONSE frame, the reply to a
* [PathChallengeFrame]. Carries the EXACT same 8-byte payload back.
* The challenger uses byte-equality to match a response to its
* outstanding challenge — a peer that echoes random bytes would
* pass validation, so callers that issue PATH_CHALLENGE MUST use
* a cryptographically-random payload.
*/
class PathResponseFrame(
val data: ByteArray,
) : Frame() {
init {
require(data.size == 8) { "PATH_RESPONSE data must be exactly 8 bytes per RFC 9000 §19.18" }
}
override fun encode(out: QuicWriter) {
out.writeByte(FrameType.PATH_RESPONSE.toInt())
out.writeBytes(data)
}
}
/**
* Decode a stream of frames from [data]. Padding bytes (0x00) are silently
* absorbed. Unknown frame types raise [QuicCodecException] (per RFC 9000 §19
@@ -445,11 +493,11 @@ fun decodeFrames(data: ByteArray): List<Frame> {
}
type == FrameType.PATH_CHALLENGE -> {
r.readBytes(8)
out += PathChallengeFrame(r.readBytes(8))
}
type == FrameType.PATH_RESPONSE -> {
r.readBytes(8)
out += PathResponseFrame(r.readBytes(8))
}
type == FrameType.CONNECTION_CLOSE_TRANSPORT -> {
@@ -113,6 +113,64 @@ class QuicStream(
val isClosed: Boolean
get() = send.finSent && receive.finReceived
/**
* True once both directions are *fully* settled and the stream may be
* removed from the connection's tracking lists (see
* `QuicConnection.retireFullyDoneStreamsLocked`). This is strictly
* stronger than [isClosed]: the latter only requires the FIN bits to
* have been observed, not that the peer has acknowledged our FIN /
* RESET (send side) nor that the application has drained the buffered
* receive bytes (receive side).
*
* Lifetime contract:
* - SEND side done: peer has ACK'd our FIN ([SendBuffer.finAcked]) OR
* we ABORTed the stream and the peer ACK'd our RESET_STREAM
* ([resetAcked]). For [Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL]
* there is no send side, so this leg is trivially done.
* - RECEIVE side done: peer FIN'd ([ReceiveBuffer.finReceived]) AND
* every byte has been delivered from the receive buffer to the
* application's incoming Channel ([ReceiveBuffer.isFullyRead]).
* Once both hold, the parser has already invoked [closeIncoming]
* so any application coroutine still draining the buffered
* [incoming] flow will terminate naturally — the QuicStream object
* can outlive its membership in the connection's `streamsList`.
* For [Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE] there is no
* receive side, so this leg is trivially done.
*
* The motivation is the moq-lite audio-rooms path: each Opus frame is
* forwarded as a fresh peer-uni stream by the relay. A 3-hour session
* at ~50 frames/sec churns ~540 000 streams. Without retirement,
* `streamsList` and the `streams` map grow monotonically — the
* `nestsClient/plans/2026-04-26-moq-lite-gap.md` soak target wants
* memory flat past handshake-stable. Stream retirement is the only
* QUIC-level fix that keeps the tracker bounded for that workload.
*
* Read this from any thread under the connection's `streamsLock` — the
* underlying flags are `@Volatile` and the buffers' synchronized
* blocks publish their state atomically.
*/
val isFullyRetired: Boolean
get() {
val sendSettled =
when (direction) {
Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL -> true
Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE,
Direction.BIDIRECTIONAL,
-> send.finAcked || resetAcked
}
if (!sendSettled) return false
val recvSettled =
when (direction) {
Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE -> true
Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL,
Direction.BIDIRECTIONAL,
-> receive.finReceived && receive.isFullyRead()
}
return recvSettled
}
/**
* Pushes [data] toward the consumer. Returns false if the bounded channel
* was full; the caller (parser) is expected to escalate to a connection-
@@ -0,0 +1,317 @@
/*
* 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.frame.AckFrame
import com.vitorpamplona.quic.frame.StreamFrame
import com.vitorpamplona.quic.stream.StreamId
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
/**
* Close-under-load races for the QuicConnection — pinning that a
* `close()` racing against a heavy stream-traffic burst doesn't
* deadlock, lose data the application already saw, or leave Flow
* collectors hanging.
*
* The motivation is the audio-rooms session-cycle pattern: a user
* joins, briefly talks (or listens), then leaves. The leave path
* fires `close()` while the moq-lite peer-uni stream churn is still
* in progress — sometimes hundreds of streams are mid-flight on the
* connection. Idle-driver close (covered by
* [QuicConnectionDriverLifecycleTest.closeIsIdempotentAndDriverJobCompletes])
* is the easy case; this file pins the load case.
*
* Three angles:
*
* 1. [closeWhileBulkStreamRetirementIsRunning] — server ACKs 100
* in-flight client-bidi streams in a single shot, triggering a
* mass retirement on the next drain. We fire `close()` while
* the retirement / writer drain is still in flight. Status MUST
* flip to CLOSED, every stream's incoming Flow MUST terminate,
* and no exception propagates to the test runner.
*
* 2. [closeWhileAppCoroutinesAreOpeningStreams] — N coroutines
* hammer `openBidiStream` while another coroutine fires
* `close()`. Every winner-of-the-race opener returns either a
* valid stream (won) or an [QuicConnectionClosedException] /
* [IllegalStateException] (lost). No coroutine hangs forever.
*
* 3. [closeWhilePeerStreamsAreInFlight] — peer is mid-stream of
* 50 server-uni group-stream payloads when close fires. Every
* incoming Flow terminates promptly with whatever bytes the
* parser had already delivered (zero or more, but never hung).
*/
class CloseUnderLoadTest {
@Test
fun closeWhileBulkStreamRetirementIsRunning() =
runBlocking {
val (client, pipe) = newConnectedClient()
val n = 100
val streams =
client.openBidiStreamsBatch(items = (0 until n).toList()) { stream, i ->
stream.send.enqueue("payload-$i".encodeToByteArray())
stream.send.finish()
stream
}
// Drain so STREAM frames go on the wire and PNs are allocated.
// After this, a server ACK covering all PNs latches finAcked
// on every stream, which makes them retire-eligible.
drainAll(client, pipe)
val largestSent = client.application.pnSpace.nextPacketNumber - 1L
assertTrue(largestSent >= 0, "drainAll must have emitted at least one app-level packet")
val ackPacket =
pipe.buildServerApplicationDatagram(
listOf(
AckFrame(
largestAcknowledged = largestSent,
ackDelay = 0L,
firstAckRange = largestSent,
),
),
)!!
feedDatagram(client, ackPacket, nowMillis = 0L)
// Now every stream is fully-retire-eligible. The next drain
// would retire them all in a single pass. Fire close()
// CONCURRENTLY with that drain — the retire path mutates
// streamsList while close()'s closeAllSignals iterates it.
// The original closeAllSignals snapshot-iterates safely, but
// a careless edit there could ConcurrentModification.
coroutineScope {
val drainer =
async { drainAll(client, pipe) }
val closer =
async {
// Tiny yield so the drainer wins the race and is
// mid-iteration when close fires. close() is
// suspending and acquires its own locks, so this
// exercises the locked-iteration ↔ locked-close
// path rather than running them strictly sequentially.
kotlinx.coroutines.yield()
client.close(0L, "close-under-load")
}
drainer.await()
closer.await()
}
// close() transitions to CLOSING; the writer's next
// drainOutbound builds a CONNECTION_CLOSE and flips to
// CLOSED. Production drives this in the driver's send loop;
// here we drive it explicitly.
drainOutbound(client, nowMillis = 0L)
assertEquals(
QuicConnection.Status.CLOSED,
client.status,
"status must flip to CLOSED after close() — even racing a retirement pass",
)
// Every stream's incoming Flow must terminate (closeAllSignals
// closes the per-stream channel, OR retirement closed it
// earlier — both end with the Flow completing). If either
// path leaks, the toList collector below times out.
val collected =
coroutineScope {
streams
.mapIndexed { idx, s ->
async {
withTimeoutOrNull(2_000L) { s.incoming.toList() } to idx
}
}.awaitAll()
}
val hung = collected.firstOrNull { it.first == null }
assertTrue(
hung == null,
"stream index ${hung?.second} incoming Flow leaked — closeAllSignals " +
"must terminate every per-stream channel even under retirement race",
)
}
@Test
fun closeWhileAppCoroutinesAreOpeningStreamsDoesNotDeadlock() =
runBlocking {
val (client, _) = newConnectedClient()
val n = 80
// openBidiStream takes streamsLock; close() takes
// lifecycleLock. The two are intentionally distinct so
// app code (open/enqueue) doesn't fight the close path.
// This test pins the lock-ordering invariant: many
// parallel openers MUST NOT deadlock against a
// concurrent close(). Whether the race actually fires
// under runBlocking's cooperative dispatcher varies
// (single-thread event loop tends to run all queued
// coroutines before the next scheduling point), so we
// don't assert on the race outcome — only on the
// absence of deadlock and the final closed state. A
// multi-threaded chaos test would exercise the race
// shape itself; the lock-ordering invariant landing
// here is the production-safety bar.
val outcome = ArrayList<Outcome>()
outcome.ensureCapacity(n)
val resultsLock = Any()
coroutineScope {
val openers =
(0 until n).map { i ->
async {
val r =
try {
client.openBidiStream()
Outcome.OPENED
} catch (_: QuicConnectionClosedException) {
Outcome.CLOSED_DURING_OPEN
} catch (_: IllegalStateException) {
Outcome.CLOSED_DURING_OPEN
} catch (_: QuicStreamLimitException) {
// Hit the peer's stream cap before close
// landed — fine, neither a hang nor a
// protocol violation.
Outcome.STREAM_LIMIT
}
synchronized(resultsLock) { outcome += r }
}
}
// Yield once so a few openers get past the lock
// acquisition before close fires, exercising the
// racing path rather than the all-after-close path.
kotlinx.coroutines.yield()
async { client.close(0L, "racing openers") }.await()
openers.awaitAll()
}
// Drain the CONNECTION_CLOSE the writer now has queued so
// status flips from CLOSING to CLOSED (see test #1 for
// rationale).
drainOutbound(client, nowMillis = 0L)
assertEquals(
QuicConnection.Status.CLOSED,
client.status,
"status must be CLOSED after close()",
)
assertEquals(
n,
outcome.size,
"every opener coroutine must have completed (no hangs) — pins the " +
"lock-ordering invariant: streamsLock (openers) and lifecycleLock " +
"(close) don't fight",
)
}
@Test
fun closeWhilePeerStreamsAreInFlight() =
runBlocking {
val (client, pipe) = newConnectedClient()
val n = 50
val firstId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
val streamIds = (0 until n).map { firstId + 4L * it }
// Half the peer streams arrive WITHOUT FIN — leaving them
// mid-flight on the connection. The other half FIN normally.
for ((idx, id) in streamIds.withIndex()) {
val payload = "frame-$idx".encodeToByteArray()
val fin = idx % 2 == 1
val packet =
pipe.buildServerApplicationDatagram(
listOf(StreamFrame(streamId = id, offset = 0L, data = payload, fin = fin)),
)!!
feedDatagram(client, packet, nowMillis = 0L)
}
// Don't drain — leaves the FIN'd streams un-retired and the
// others mid-flight. close() fires while every stream is in
// some live state.
client.close(0L, "close mid peer-stream burst")
// Drain so the writer emits CONNECTION_CLOSE and status
// moves CLOSING → CLOSED. Without this drain, a test on
// the in-memory pipe would observe CLOSING; the production
// driver fires drainOutbound automatically.
drainOutbound(client, nowMillis = 0L)
assertEquals(QuicConnection.Status.CLOSED, client.status)
// Every stream's incoming Flow must terminate (closeAllSignals
// ran). The half that got FIN return their bytes; the other
// half terminate with whatever was buffered (may be 0 bytes).
val collected =
coroutineScope {
streamIds
.map { id ->
async {
val s = client.streamById(id)
if (s == null) {
null to id // never created, treat as "no leak"
} else {
withTimeoutOrNull(2_000L) { s.incoming.toList() } to id
}
}
}.awaitAll()
}
val hung = collected.firstOrNull { it.first == null && it.second != -1L && client.streamByIdLockedForTest(it.second) != null }
assertTrue(
hung == null,
"stream id ${hung?.second} incoming Flow leaked across close — closeAllSignals " +
"must terminate every live stream's channel",
)
}
private enum class Outcome {
OPENED,
CLOSED_DURING_OPEN,
STREAM_LIMIT,
}
private fun drainAll(
client: QuicConnection,
pipe: InMemoryQuicPipe,
) {
while (true) {
val out = drainOutbound(client, nowMillis = 0L) ?: break
pipe.decryptClientApplicationFrames(out)
}
}
// 1024-stream caps so the 100-bidi-stream open burst doesn't
// brush the cap mid-test, and 16 MiB data window so multi-payload
// traffic doesn't trip flow-control mid-close.
private fun newConnectedClient(): Pair<QuicConnection, InMemoryQuicPipe> =
com.vitorpamplona.quic.connection.newConnectedClient(
maxStreamsBidi = 1024,
maxStreamsUni = 1024,
maxData = 16L * 1024 * 1024,
)
}
/**
* Test-only synchronous lookup that avoids the suspending
* [QuicConnection.streamById] for use inside `firstOrNull`. Caller
* doesn't need the lock — this is a best-effort post-close check;
* the streams map is no longer being mutated by the time
* `closeAllSignals` returns.
*/
private fun QuicConnection.streamByIdLockedForTest(id: Long) = streamsListLocked().firstOrNull { it.streamId == id }
@@ -0,0 +1,94 @@
/*
* 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.tls.InProcessTlsServer
import com.vitorpamplona.quic.tls.PermissiveCertificateValidator
import kotlinx.coroutines.runBlocking
import kotlin.test.assertEquals
/**
* Stand up a fresh [QuicConnection] wired through an
* [InMemoryQuicPipe] and drive the handshake to CONNECTED. Most of
* the audio-rooms tests start from this exact shape — extracted
* here to keep each test file focused on its own assertions
* rather than ~40 lines of identical fixture boilerplate.
*
* The transport-parameter knobs cover the only variation real tests
* need:
* - moq-lite-shaped tests (many peer-uni streams, large data
* window): pass `maxStreamsUni = 65_536`, `maxData = 16 MiB`.
* - single-stream / control-frame tests (defaults are plenty).
*
* Both client and server advertise the same caps so flow-control
* regressions surface as caps-mismatch failures rather than tests
* passing accidentally because ONE side was generous.
*/
fun newConnectedClient(
serverName: String = "example.test",
maxStreamsBidi: Long = 16,
maxStreamsUni: Long = 16,
maxData: Long = 1L * 1024 * 1024,
maxStreamData: Long = 64L * 1024,
handshakeRounds: Int = 16,
): Pair<QuicConnection, InMemoryQuicPipe> =
runBlocking {
val client =
QuicConnection(
serverName = serverName,
config =
QuicConnectionConfig(
initialMaxStreamsBidi = maxStreamsBidi,
initialMaxStreamsUni = maxStreamsUni,
initialMaxData = maxData,
initialMaxStreamDataBidiLocal = maxStreamData,
initialMaxStreamDataBidiRemote = maxStreamData,
initialMaxStreamDataUni = maxStreamData,
),
tlsCertificateValidator = PermissiveCertificateValidator(),
)
val serverScid = ConnectionId.random(8)
val tlsServer =
InProcessTlsServer(
transportParameters =
TransportParameters(
initialMaxData = maxData,
initialMaxStreamDataBidiLocal = maxStreamData,
initialMaxStreamDataBidiRemote = maxStreamData,
initialMaxStreamDataUni = maxStreamData,
initialMaxStreamsBidi = maxStreamsBidi,
initialMaxStreamsUni = maxStreamsUni,
initialSourceConnectionId = serverScid.bytes,
originalDestinationConnectionId = client.destinationConnectionId.bytes,
).encode(),
)
val pipe =
InMemoryQuicPipe(
client = client,
initialDcid = client.destinationConnectionId.bytes,
serverScid = serverScid,
tlsServer = tlsServer,
)
client.start()
pipe.drive(maxRounds = handshakeRounds)
assertEquals(QuicConnection.Status.CONNECTED, client.status)
client to pipe
}
@@ -24,6 +24,7 @@ import com.vitorpamplona.quic.crypto.Aes128Gcm
import com.vitorpamplona.quic.crypto.AesEcbHeaderProtection
import com.vitorpamplona.quic.crypto.InitialSecrets
import com.vitorpamplona.quic.crypto.PlatformAesOneBlock
import com.vitorpamplona.quic.crypto.expandLabel
import com.vitorpamplona.quic.frame.AckFrame
import com.vitorpamplona.quic.frame.ConnectionCloseFrame
import com.vitorpamplona.quic.frame.CryptoFrame
@@ -97,6 +98,42 @@ class InMemoryQuicPipe(
private var serverApplicationRx: PacketProtection? = null
private var serverApplicationTx: PacketProtection? = null
/**
* Mirror of the client's `application.sendProtection` for the
* pipe's TX side. The key-update test rotates this forward
* (HKDF-Expand-Label "quic ku") so the pipe can emit a packet
* encrypted with the next-phase keys, exercising the client's
* peer-initiated rotation path. Stays null until the handshake
* completes; updated in lockstep with [serverApplicationTx]
* by [installServerSecretsAfterHandshakeBegin] and rolled
* forward by [rotateServerApplicationKeys].
*/
private var serverSendApplicationSecret: ByteArray? = null
/**
* Cipher suite negotiated by [tlsServer]. Cached after the
* handshake completes so [rotateServerApplicationKeys] doesn't
* have to re-query the TLS object on every rotation.
*/
private var serverApplicationCipherSuite: Int = 0
/**
* Stashed pre-rotation TX keys, so the test can re-emit a
* "reordered" packet on the OLD keys after a rotation has
* committed. Without this, the reorder-window check (RFC 9001
* §6.1: client retains [previousReceiveProtection] until the
* reorder window closes) would have nothing to exercise.
* Updated by [rotateServerApplicationKeys].
*/
private var serverApplicationTxPrior: PacketProtection? = null
/**
* Tracks the server's send-side key-phase bit. Flipped by
* [rotateServerApplicationKeys] so [buildServerApplicationDatagram]
* can stamp the right value into the short-header.
*/
private var serverSendKeyPhase: Boolean = false
private val initialPnSpace = PacketNumberSpaceState()
private val handshakePnSpace = PacketNumberSpaceState()
private val applicationPnSpace = PacketNumberSpaceState()
@@ -226,6 +263,8 @@ class InMemoryQuicPipe(
serverHandshakeTx = packetProtectionFromSecret(cipher, tlsServer.serverHandshakeSecret!!)
serverApplicationRx = packetProtectionFromSecret(cipher, tlsServer.clientApplicationSecret!!)
serverApplicationTx = packetProtectionFromSecret(cipher, tlsServer.serverApplicationSecret!!)
serverSendApplicationSecret = tlsServer.serverApplicationSecret
serverApplicationCipherSuite = cipher
}
/** Build a single datagram from the server containing whatever it owes the client. */
@@ -293,6 +332,87 @@ class InMemoryQuicPipe(
dcid = client.sourceConnectionId,
packetNumber = pn,
payload = payload,
keyPhase = serverSendKeyPhase,
),
proto.aead,
proto.key,
proto.iv,
proto.hp,
proto.hpKey,
largestAckedInSpace = applicationPnSpace.largestReceived,
)
}
/**
* Test-only: roll the server's TX-side application keys forward by
* one phase (RFC 9001 §6.1) and stash the prior keys on
* [serverApplicationTxPrior] so a follow-up call to
* [buildServerApplicationDatagramWithPriorKeys] can simulate a
* reordered packet from before the rotation.
*
* After this call, the next [buildServerApplicationDatagram] will
* emit a packet whose KEY_PHASE bit is the rotated value and
* whose AEAD nonce/key are HKDF-derived off the next-phase
* secret. The pipe also tracks its own send-phase bit, so we
* model the realistic case where the server (peer) rotates its
* own send keys and the client mirrors on receive.
*
* Throws if called before the handshake completes.
*/
fun rotateServerApplicationKeys() {
val curSecret =
checkNotNull(serverSendApplicationSecret) {
"rotateServerApplicationKeys called before handshake completed"
}
val cs = serverApplicationCipherSuite
check(cs != 0) { "negotiated cipher suite not yet recorded" }
val nextSecret = expandLabel(curSecret, "quic ku", curSecret.size)
val nextProto = packetProtectionFromSecret(cipherSuite = cs, secret = nextSecret)
val live =
checkNotNull(serverApplicationTx) {
"rotateServerApplicationKeys called before serverApplicationTx installed"
}
// RFC 9001 §6.1 — HP key is NOT updated on rotation; carry the
// existing one forward.
val rotated =
PacketProtection(
aead = nextProto.aead,
key = nextProto.key,
iv = nextProto.iv,
hp = live.hp,
hpKey = live.hpKey,
)
serverApplicationTxPrior = live
serverApplicationTx = rotated
serverSendApplicationSecret = nextSecret
serverSendKeyPhase = !serverSendKeyPhase
}
/**
* Test-only: build a server application packet using the
* pre-rotation keys captured by [rotateServerApplicationKeys].
* Used to drive the reorder-window code path in the parser
* ([com.vitorpamplona.quic.connection.feedShortHeaderPacket]'s
* `previousReceiveProtection != null` branch) — the client
* MUST decrypt this packet with [QuicConnection.previousReceiveProtection]
* even after committing the rotation, because in the real network
* a reordered packet from before the rotation can arrive after
* the rotation-triggering packet.
*
* Returns null if no prior keys have been stashed (i.e.
* [rotateServerApplicationKeys] hasn't been called yet).
*/
fun buildServerApplicationDatagramWithPriorKeys(frames: List<com.vitorpamplona.quic.frame.Frame>): ByteArray? {
val proto = serverApplicationTxPrior ?: return null
val pn = applicationPnSpace.allocateOutbound()
val payload = encodeFrames(frames)
return ShortHeaderPacket.build(
com.vitorpamplona.quic.packet.ShortHeaderPlaintextPacket(
dcid = client.sourceConnectionId,
packetNumber = pn,
payload = payload,
// Prior phase is the inverse of the current one.
keyPhase = !serverSendKeyPhase,
),
proto.aead,
proto.key,
@@ -0,0 +1,362 @@
/*
* 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.frame.PingFrame
import com.vitorpamplona.quic.frame.StreamFrame
import com.vitorpamplona.quic.packet.ShortHeaderPacket
import com.vitorpamplona.quic.stream.StreamId
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertNotNull
import kotlin.test.assertTrue
/**
* Pins the peer-initiated 1-RTT key update path (RFC 9001 §6).
*
* Soak target #2 from the audio-rooms hardening pass: long-lived
* audio-room sessions outlive the QUIC AEAD key's safe usage window
* (~2^60 packets per RFC 9001 §6.6, but real implementations rotate
* far earlier quic-go defaults to ~100 packets, picoquic to a few
* thousand). The client MUST handle a peer that flips the
* `KEY_PHASE` bit by deriving next-phase keys, retry-decrypting the
* triggering packet, and then mirroring the rotation on its own
* send side. Without this, post-rotation packets AEAD-fail
* silently qlog shows them as "AEAD auth failed" drops, the
* connection wedges (no ACKs to peer, peer falls into PTO mode), and
* audio cuts off.
*
* The implementation lives across:
* - `QuicConnectionParser.feedShortHeaderPacket` (peek key-phase
* bit, route to current/previous/next keys).
* - `QuicConnection.deriveNextPhaseReceiveKeys` (HKDF-Expand-Label
* "quic ku" next secret key/iv).
* - `QuicConnection.commitKeyUpdate` (install next as live, demote
* live to previous, mirror onto send side, flip phase bits).
* - `QuicConnectionWriter` stamping `currentSendKeyPhase` into the
* short header on each outbound build.
*
* These tests exercise the peer-initiated path end-to-end against
* `InMemoryQuicPipe`, which now grows a `rotateServerApplicationKeys`
* helper that walks the same HKDF dance the production peer would.
*/
class KeyUpdatePeerInitiatedTest {
@Test
fun peerInitiatedRotationCommitsAndMirrorsOnSend() =
runBlocking {
val (client, pipe) = newConnectedClient()
// Pre-rotation invariants — pin the baseline so a regression
// that fired commitKeyUpdate during the handshake itself
// (it shouldn't) would surface here as a phase mismatch.
assertEquals(false, client.currentReceiveKeyPhase, "key-phase 0 at start")
assertEquals(false, client.currentSendKeyPhase, "send-phase 0 at start")
assertEquals(null, client.previousReceiveProtection, "no prior keys before rotation")
val originalReceiveProtection = client.application.receiveProtection
assertNotNull(originalReceiveProtection, "client must have 1-RTT receive keys after handshake")
val originalSendProtection = client.application.sendProtection
assertNotNull(originalSendProtection, "client must have 1-RTT send keys after handshake")
// Rotate the pipe's TX keys forward and emit a packet whose
// body is a plain PING — small payload, ack-eliciting so the
// client also has a reason to send back. The packet's
// KEY_PHASE bit is true.
pipe.rotateServerApplicationKeys()
val rotatedPacket = pipe.buildServerApplicationDatagram(listOf(PingFrame))!!
// Sanity: the bit on the wire is the rotated phase.
assertEquals(
true,
peekKeyPhase(rotatedPacket, client),
"the test fixture must produce a KEY_PHASE=1 packet after rotateServerApplicationKeys",
)
feedDatagram(client, rotatedPacket, nowMillis = 0L)
// Post-rotation invariants:
// - currentReceiveKeyPhase flipped (commitKeyUpdate ran).
// - currentSendKeyPhase flipped in lockstep (we mirror).
// - previousReceiveProtection holds the pre-rotation keys
// (kept alive for the reorder window).
// - application.receiveProtection is a NEW instance (next-
// phase keys derived from HKDF "quic ku").
// - application.sendProtection is also a NEW instance
// (mirror onto send side).
// - connection stays CONNECTED — the rotation isn't a
// teardown trigger.
assertEquals(
true,
client.currentReceiveKeyPhase,
"currentReceiveKeyPhase must flip after the parser commits the rotation",
)
assertEquals(
true,
client.currentSendKeyPhase,
"currentSendKeyPhase must mirror the receive-side rotation in lockstep " +
"(commitKeyUpdate on the send side derives the matching next-phase secret)",
)
assertEquals(
originalReceiveProtection,
client.previousReceiveProtection,
"pre-rotation receive keys must be retained as previousReceiveProtection " +
"for the reorder window (RFC 9001 §6.1)",
)
assertTrue(
client.application.receiveProtection !== originalReceiveProtection,
"application.receiveProtection must reference the next-phase keys, not the originals",
)
assertTrue(
client.application.sendProtection !== originalSendProtection,
"application.sendProtection must reference the next-phase keys after the mirror",
)
assertEquals(
QuicConnection.Status.CONNECTED,
client.status,
"connection must stay CONNECTED across a peer-initiated key update",
)
}
@Test
fun reorderedPacketOnPriorKeysStillDecryptsAfterRotation() =
runBlocking {
// RFC 9001 §6.1 reorder window: once the client has rotated,
// it MUST keep the prior-phase receive keys for some bounded
// window so that a packet sent before the peer rotated (but
// delayed in the network) still decrypts. The parser path is
// the `previousReceiveProtection != null` arm in
// feedShortHeaderPacket. Pin it by:
// 1. Pushing one peer-uni stream + FIN with the pre-rotation
// keys to establish that the client has live state on
// stream id 3 (server-uni #0).
// 2. Rotating, pushing a rotation-triggering PING.
// 3. Replaying a SECOND payload on the same stream id with
// the PRIOR keys — i.e. the pipe re-uses the stashed
// pre-rotation TX. The client must decrypt it via
// previousReceiveProtection and surface the bytes.
val (client, pipe) = newConnectedClient()
val streamId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
// Pre-rotation: bytes 0..3 with no FIN.
val prePayload = "pre-".encodeToByteArray()
val prePacket =
pipe.buildServerApplicationDatagram(
listOf(StreamFrame(streamId = streamId, offset = 0L, data = prePayload, fin = false)),
)!!
feedDatagram(client, prePacket, nowMillis = 0L)
// Rotate, then trigger commit with a PING.
pipe.rotateServerApplicationKeys()
feedDatagram(client, pipe.buildServerApplicationDatagram(listOf(PingFrame))!!, nowMillis = 0L)
assertEquals(true, client.currentReceiveKeyPhase, "rotation must commit before reorder test")
assertNotNull(client.previousReceiveProtection, "prior keys must be retained")
// Now the reorder: the peer SENT this packet pre-rotation
// but it arrives after the rotation-triggering PING. The
// pipe builds it with the stashed prior keys. The packet's
// KEY_PHASE bit is the pre-rotation value (false), and the
// client's parser MUST route through previousReceiveProtection.
val reorderedPayload = "post".encodeToByteArray()
val reorderedPacket =
pipe.buildServerApplicationDatagramWithPriorKeys(
listOf(
StreamFrame(
streamId = streamId,
offset = prePayload.size.toLong(),
data = reorderedPayload,
fin = true,
),
),
)!!
assertEquals(
false,
peekKeyPhase(reorderedPacket, client),
"reordered packet must carry the pre-rotation KEY_PHASE bit",
)
feedDatagram(client, reorderedPacket, nowMillis = 0L)
// The application sees the FULL stream payload despite the
// mid-stream key rotation. If the parser had dropped the
// reordered packet, the stream would stall (no FIN) and the
// toList collector below would time out.
val stream = client.streamById(streamId)!!
val chunks = withTimeoutOrNull(2_000L) { stream.incoming.toList() }
assertNotNull(chunks, "stream incoming Flow must complete despite the mid-stream rotation")
val joined = ByteArray(chunks.sumOf { it.size })
var p = 0
for (c in chunks) {
c.copyInto(joined, p)
p += c.size
}
assertEquals(
"pre-post",
joined.decodeToString(),
"reordered packet decrypted on prior keys must surface its bytes intact",
)
assertEquals(
QuicConnection.Status.CONNECTED,
client.status,
"connection must stay CONNECTED across the reorder",
)
}
@Test
fun postRotationOutboundPacketCarriesNewKeyPhaseAndDecryptsForPeer() =
runBlocking {
// Send-side correctness check: after a peer-initiated
// rotation, the writer's NEXT outbound packet must
// (a) stamp the new currentSendKeyPhase into the
// short header, and
// (b) be encrypted with the rotated send keys (so the
// peer, which has also rotated, decrypts it
// correctly).
// If commitKeyUpdate's send-side mirror regressed (e.g.
// updated currentSendKeyPhase but forgot to install fresh
// sendProtection), the wire bit and the AEAD keys would
// disagree and the peer would drop the packet — visible as
// a silent "client never ACKs after rotation" wedge.
val (client, pipe) = newConnectedClient()
pipe.rotateServerApplicationKeys()
feedDatagram(client, pipe.buildServerApplicationDatagram(listOf(PingFrame))!!, nowMillis = 0L)
assertEquals(true, client.currentSendKeyPhase, "rotation must commit before send-side check")
// The PING is ack-eliciting, so the writer has work to do.
// drainOutbound builds the ACK + any other queued frames.
val outbound = drainOutbound(client, nowMillis = 0L)
assertNotNull(outbound, "writer must emit an ACK in response to the rotation-trigger PING")
// Pipe re-uses the same applicationPnSpace and HP key, so
// it can decrypt this packet via decryptClientApplicationFrames
// — but only if the AEAD keys match. The pipe's RX side
// ALSO has to rotate to match; the existing pipe
// doesn't rotate RX automatically, so the AEAD will fail
// and we just check the wire-level bit through peekKeyPhase.
// What we CAN observe without rotating the pipe RX is the
// unprotected first byte: header-protection key isn't
// rotated (RFC 9001 §6.1), so the HP unmask still works
// and tells us the wire bit.
val wirePhase = peekKeyPhase(outbound, client, useSendKeys = true)
assertEquals(
true,
wirePhase,
"post-rotation outbound packet must carry KEY_PHASE=1 on the wire — the writer reads " +
"currentSendKeyPhase (now true) when stamping the short header",
)
assertEquals(QuicConnection.Status.CONNECTED, client.status)
}
@Test
fun twoConsecutiveRotationsCommitCorrectly() =
runBlocking {
// Belt + braces: one rotation is the simple case; a second
// rotation must derive off the FIRST-rotation secret, not
// the original handshake secret, and the parser must NOT
// misroute the second-rotation packet to
// previousReceiveProtection (which would AEAD-fail and
// silently drop, wedging the connection — the
// KNOWN-LIMITATION pre-fix this test pins now closes).
//
// The parser fix is "try previous keys; on AEAD failure
// fall through to next-phase derivation" — neqo's
// approach. Two AEAD attempts on a mismatched-phase
// packet are cheap; KEY_PHASE mismatch is rare to begin
// with (at most once per a-billion-packets in normal
// usage).
val (client, pipe) = newConnectedClient()
pipe.rotateServerApplicationKeys()
feedDatagram(client, pipe.buildServerApplicationDatagram(listOf(PingFrame))!!, nowMillis = 0L)
assertEquals(true, client.currentReceiveKeyPhase, "first rotation must flip the bit")
pipe.rotateServerApplicationKeys()
feedDatagram(client, pipe.buildServerApplicationDatagram(listOf(PingFrame))!!, nowMillis = 0L)
assertEquals(
false,
client.currentReceiveKeyPhase,
"second rotation must flip the bit back to 0 — failure here means the parser " +
"misrouted the second rotation through previousReceiveProtection (the prior " +
"key-update bug) instead of falling through to next-phase derivation",
)
assertEquals(
false,
client.currentSendKeyPhase,
"send-side mirror must also have rolled forward through both rotations",
)
assertEquals(QuicConnection.Status.CONNECTED, client.status)
}
/**
* Read the unprotected KEY_PHASE bit out of a packet so the test
* can assert the wire shape independent of the client's
* bookkeeping.
*
* Walks past any leading long-header (Initial / Handshake) packets
* in the datagram drainOutbound will keep emitting a Handshake-
* level ACK at the front of each datagram until the InMemoryQuicPipe
* delivers a HANDSHAKE_DONE frame (which it doesn't), so the short-
* header packet is rarely first on the wire.
*
* For inbound (server-built) packets we hand the client's RECEIVE
* HP key. For outbound (client-built) packets we hand the client's
* SEND HP key. Both phases share their direction's HP key per RFC
* 9001 §6.1 ("the header_protection key is not updated when keys
* are updated"), so this stays valid across rotations.
*/
private fun peekKeyPhase(
packet: ByteArray,
client: QuicConnection,
useSendKeys: Boolean = false,
): Boolean? {
val live =
if (useSendKeys) {
client.application.sendProtection
} else {
client.application.receiveProtection
} ?: return null
var offset = 0
while (offset < packet.size) {
val first = packet[offset].toInt() and 0xFF
if ((first and 0x80) == 0) {
// Short header — what we want.
return ShortHeaderPacket
.peekKeyPhase(
bytes = packet,
offset = offset,
dcidLen = client.sourceConnectionId.length,
hp = live.hp,
hpKey = live.hpKey,
)?.keyPhase
}
// Long header — skip past using the encoded length field.
val peeked =
com.vitorpamplona.quic.packet.LongHeaderPacket
.peekHeader(packet, offset) ?: return null
offset += peeked.totalLength
}
return null
}
// Default caps (16/16 streams, 1 MiB data) — the rotation tests
// don't push much traffic; small caps keep the pipe handshake fast.
private fun newConnectedClient(): Pair<QuicConnection, InMemoryQuicPipe> =
com.vitorpamplona.quic.connection
.newConnectedClient()
}
@@ -0,0 +1,401 @@
/*
* 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.frame.StreamFrame
import com.vitorpamplona.quic.stream.StreamId
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import kotlin.random.Random
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertNotNull
import kotlin.test.assertTrue
/**
* First pass at soak target #5 (transient packet loss + jitter) for
* the moq-lite group-stream shape.
*
* **Scope of this draft.** moq-lite's group streams are best-effort
* (the [SendBuffer.bestEffort] flag drops lost ranges instead of
* retransmitting). On the LISTENER side that means: a peer-uni
* stream that the server originated, whose carrying datagram never
* arrives, is gone the relay does not retransmit it. The listener
* application is expected to observe a small fraction of frames
* "missing" but the connection itself stays healthy and audio for
* the surviving frames continues.
*
* This test pins exactly that contract:
* 1. Server publishes 50 group streams (one StreamFrame + FIN per
* stream, one stream per datagram the moq-lite group shape
* 1:1).
* 2. A configurable loss model drops 5% (uniformly random) of
* server datagrams before [feedDatagram].
* 3. We assert the listener surfaces 90% of the published
* streams ( 45 of 50 at p_loss=0.05) and the connection stays
* CONNECTED.
*
* What's intentionally OUT of scope here:
* - **Reordering / jitter.** moq-lite tolerates reordering by
* construction (each group is its own stream, contiguous within
* itself). A reorder-injecting wrapper is the obvious next
* layer; this draft pins the loss-only contract first.
* - **Reliable bidi STREAM frame retransmit on loss.** Covered by
* the existing CryptoRetransmitTest / MultiplexingRoundTripTest;
* the transferloss / handshakeloss interop testcases drive that
* end-to-end.
* - **Latency under loss.** Worth measuring once we wire a real
* moq-lite publisher; for now `runBlocking` + in-memory pipe
* means "real time" is meaningless.
*
* Production link to chase next:
* - `nestsClient/.../moq/lite/Subscriber.kt` the listener path.
* Verify it doesn't tear down its own subscription on a missing
* group sequence; surfacing the gap to the audio decoder
* (silence frames) is the right behaviour.
*/
class MoqLiteLossHarnessTest {
@Test
fun listenerToleratesFivePercentLossOnGroupStreams() =
runBlocking {
// Deterministic RNG so a CI flake is reproducible: the
// exact dropped indices are seeded from a fixed value.
// Bumping the seed to find pathological loss patterns is
// a good follow-up, but for now we just need any seed
// that exercises the dropped-and-recovered path.
val rng = Random(0xC01DBEEFL)
val groupCount = 50
val lossRate = 0.05
val (client, pipe) = newConnectedClient()
val firstId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
val streamIds = (0 until groupCount).map { firstId + 4L * it }
val payloads = streamIds.associateWith { id -> "frame-$id".encodeToByteArray() }
val droppedIds = mutableSetOf<Long>()
for (id in streamIds) {
val drop = rng.nextDouble() < lossRate
if (drop) {
droppedIds += id
continue // Don't even build the packet — same shape as kernel
// dropping the datagram before it reaches the QUIC parser.
}
val frame =
StreamFrame(
streamId = id,
offset = 0L,
data = payloads[id]!!,
fin = true,
)
val packet = pipe.buildServerApplicationDatagram(listOf(frame))!!
feedDatagram(client, packet, nowMillis = 0L)
}
// Drain the listener's incoming Flow for every stream we
// EXPECTED to receive (= all server-uni IDs minus the dropped
// ones). Streams that were dropped never had a QuicStream
// created on our side, so streamById(id) returns null — we
// don't even try to drain those.
val received = mutableMapOf<Long, ByteArray>()
for (id in streamIds) {
if (id in droppedIds) continue
val stream = client.streamById(id) ?: continue
val chunks = withTimeoutOrNull(2_000L) { stream.incoming.toList() } ?: continue
val joined = ByteArray(chunks.sumOf { it.size })
var p = 0
for (c in chunks) {
c.copyInto(joined, p)
p += c.size
}
received[id] = joined
}
// Surviving streams must surface their full payload.
for ((id, bytes) in received) {
assertEquals(
payloads[id]!!.decodeToString(),
bytes.decodeToString(),
"stream $id payload corrupted under loss — best-effort path " +
"must NOT split or drop bytes WITHIN a delivered group",
)
}
// ≥ 90% delivered (lossRate=5% — band leaves room for RNG
// tail-end where the seed happens to drop more than the
// mean).
val expectedFloor = (groupCount * 0.9).toInt()
assertTrue(
received.size >= expectedFloor,
"listener received only ${received.size} / $groupCount groups under " +
"${(lossRate * 100).toInt()}% loss — floor was $expectedFloor. " +
"dropped=${droppedIds.size} ids=${droppedIds.sorted()}",
)
assertEquals(
QuicConnection.Status.CONNECTED,
client.status,
"connection must stay CONNECTED across packet loss on best-effort streams",
)
}
@Test
fun listenerSurfacesEveryFrameWhenLossRateIsZero() =
runBlocking {
// Sanity for the harness itself: with lossRate=0 we must
// see all 50 frames intact. If this fails the loss-shape
// assertions in the lossy test are unreliable.
val (client, pipe) = newConnectedClient()
val groupCount = 50
val firstId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
for (i in 0 until groupCount) {
val id = firstId + 4L * i
val payload = "frame-$id".encodeToByteArray()
val frame = StreamFrame(streamId = id, offset = 0L, data = payload, fin = true)
val packet = pipe.buildServerApplicationDatagram(listOf(frame))!!
feedDatagram(client, packet, nowMillis = 0L)
}
var delivered = 0
for (i in 0 until groupCount) {
val id = firstId + 4L * i
val stream = client.streamById(id)!!
val chunks = withTimeoutOrNull(2_000L) { stream.incoming.toList() }
if (chunks != null) delivered += 1
}
assertEquals(groupCount, delivered, "lossRate=0 baseline must deliver every frame")
}
@Test
fun listenerToleratesPacketReorderingOnGroupStreams() =
runBlocking {
// Reorder injection — the network can deliver datagrams
// out of order even when none are dropped. moq-lite group
// streams MUST tolerate this: each stream is a self-
// contained Opus frame (offset 0 + FIN), so reorder at
// the datagram level just means the listener sees
// streams arrive in a different order than the relay
// sent them. Audio frames carry sequence numbers, so the
// application-level player handles late-arrivers via its
// own jitter buffer.
//
// Pin the contract: 100% delivery under arbitrary
// datagram-order permutation. This catches a class of
// regression where a parser invariant ("PN must
// increase") gets accidentally tightened to "STREAM IDs
// must arrive in order", which would silently break
// moq-lite under any real-world jitter.
val rng = Random(0xBAD5EEDL)
val groupCount = 50
val (client, pipe) = newConnectedClient()
val firstId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
val streamIds = (0 until groupCount).map { firstId + 4L * it }
val payloads = streamIds.associateWith { id -> "frame-$id".encodeToByteArray() }
// Build all packets up-front; then shuffle the delivery
// order. Permutation drives reorder of:
// - Stream IDs (peer-uni IDs are globally ordered;
// the listener sees later IDs before earlier ones).
// - QUIC packet numbers (each datagram carries a fresh
// PN; reorder means the parser's PN-space tracking
// observes non-monotonic largestReceived advances).
val packets =
streamIds.map { id ->
pipe.buildServerApplicationDatagram(
listOf(StreamFrame(streamId = id, offset = 0L, data = payloads[id]!!, fin = true)),
)!!
}
val deliveryOrder = packets.indices.shuffled(rng)
for (idx in deliveryOrder) {
feedDatagram(client, packets[idx], nowMillis = 0L)
}
// Every stream must have surfaced its full payload
// despite arriving out of order.
for (id in streamIds) {
val s = client.streamById(id)!!
val chunks = withTimeoutOrNull(2_000L) { s.incoming.toList() }
assertNotNull(chunks, "stream $id must surface despite reorder")
val joined = ByteArray(chunks.sumOf { it.size })
var p = 0
for (c in chunks) {
c.copyInto(joined, p)
p += c.size
}
assertEquals(payloads[id]!!.decodeToString(), joined.decodeToString())
}
assertEquals(QuicConnection.Status.CONNECTED, client.status)
}
@Test
fun listenerSurvivesExtremeTwentyPercentLoss() =
runBlocking {
// Extreme-loss canary: 20% drop is far past the typical
// 1-5% range a healthy mobile network sees, but
// approaches what a degraded subway / elevator path
// delivers. moq-lite's contract here is "best-effort —
// gaps surface to audio decoder as silence." The QUIC
// contract is "the connection itself stays healthy."
//
// The audible-quality bar is the application's problem
// (jitter buffer, FEC). What this test pins is that the
// QUIC LAYER doesn't tear itself down under aggressive
// loss — flow-control accounting, ACK-tracker windows,
// and the retired-stream-id ring all stay consistent.
// A regression in any of those would either fire a
// protocol violation (CONNECTION_CLOSE) or wedge the
// peer into PTO mode.
val rng = Random(0xDEADBEEFL)
val groupCount = 200
val lossRate = 0.20
val (client, pipe) = newConnectedClient()
val firstId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
val streamIds = (0 until groupCount).map { firstId + 4L * it }
val payloads = streamIds.associateWith { id -> "frame-$id".encodeToByteArray() }
val droppedIds = mutableSetOf<Long>()
for (id in streamIds) {
if (rng.nextDouble() < lossRate) {
droppedIds += id
continue
}
val packet =
pipe.buildServerApplicationDatagram(
listOf(StreamFrame(streamId = id, offset = 0L, data = payloads[id]!!, fin = true)),
)!!
feedDatagram(client, packet, nowMillis = 0L)
}
val received = mutableListOf<Long>()
for (id in streamIds) {
if (id in droppedIds) continue
val s = client.streamById(id) ?: continue
if (withTimeoutOrNull(2_000L) { s.incoming.toList() } != null) {
received += id
}
}
// Expected delivery is approximately (1 - lossRate) of the
// total. RNG variance gives ~5% tolerance band; we assert
// ≥ 60% to leave plenty of margin while still catching a
// catastrophic collapse (e.g. parser silently drops every
// packet after the first loss).
val floor = (groupCount * 0.6).toInt()
assertTrue(
received.size >= floor,
"listener received ${received.size} / $groupCount under 20% loss — floor $floor. " +
"dropped=${droppedIds.size}",
)
assertEquals(
QuicConnection.Status.CONNECTED,
client.status,
"connection must stay CONNECTED through extreme loss",
)
}
@Test
fun reliableBidiStreamRecoversFromMidStreamPacketLoss() =
runBlocking {
// Reliable-stream loss harness — distinct from the moq-
// lite group-stream best-effort path above. RFC 9000
// STREAM frames carry full reliability semantics: a
// dropped packet must trigger retransmit on PTO, and
// the listener MUST eventually see the bytes contiguous
// and in order. This test pins that contract end-to-end
// by:
// 1. Server sends 4 STREAM frames on the same bidi
// stream, each in its own datagram, covering a
// monotonic offset range.
// 2. The middle two datagrams are dropped on first
// delivery — simulating a transient mid-stream
// loss.
// 3. Server retransmits the dropped ranges on the
// same offsets (mimicking what RFC 9002
// retransmit logic does on PTO).
// 4. Client must surface all 4 chunks contiguous,
// in order, no gaps.
val (client, pipe) = newConnectedClient()
// Open a client-bidi stream so the server can write back.
val stream = client.openBidiStream()
val streamId = stream.streamId
val chunks =
listOf(
"AAAA".encodeToByteArray(),
"BBBB".encodeToByteArray(),
"CCCC".encodeToByteArray(),
"DDDD".encodeToByteArray(),
)
val offsets = LongArray(chunks.size)
var off = 0L
for ((i, c) in chunks.withIndex()) {
offsets[i] = off
off += c.size
}
// First-pass delivery: drop chunks[1] and chunks[2].
for (i in chunks.indices) {
if (i == 1 || i == 2) continue
val frame =
StreamFrame(
streamId = streamId,
offset = offsets[i],
data = chunks[i],
fin = i == chunks.size - 1,
)
feedDatagram(client, pipe.buildServerApplicationDatagram(listOf(frame))!!, nowMillis = 0L)
}
// Retransmit the dropped chunks on their original offsets.
for (i in listOf(1, 2)) {
val frame =
StreamFrame(
streamId = streamId,
offset = offsets[i],
data = chunks[i],
fin = false,
)
feedDatagram(client, pipe.buildServerApplicationDatagram(listOf(frame))!!, nowMillis = 0L)
}
val collected = withTimeoutOrNull(2_000L) { stream.incoming.toList() }
assertNotNull(collected, "stream must complete after retransmits fill the gaps")
val joined = ByteArray(collected.sumOf { it.size })
var p = 0
for (c in collected) {
c.copyInto(joined, p)
p += c.size
}
assertEquals(
"AAAABBBBCCCCDDDD",
joined.decodeToString(),
"reliable bidi stream must surface every byte in offset order even after " +
"mid-stream packet loss + retransmit",
)
assertEquals(QuicConnection.Status.CONNECTED, client.status)
}
// 1024 streams per direction + 16 MiB connection-level data —
// headroom for 200-stream loss tests without bumping caps.
private fun newConnectedClient(): Pair<QuicConnection, InMemoryQuicPipe> =
com.vitorpamplona.quic.connection.newConnectedClient(
maxStreamsBidi = 1024,
maxStreamsUni = 1024,
maxData = 16L * 1024 * 1024,
)
}
@@ -0,0 +1,217 @@
/*
* 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.frame.PathChallengeFrame
import com.vitorpamplona.quic.frame.PathResponseFrame
import com.vitorpamplona.quic.frame.decodeFrames
import com.vitorpamplona.quic.frame.encodeFrames
import kotlinx.coroutines.runBlocking
import kotlin.test.Test
import kotlin.test.assertContentEquals
import kotlin.test.assertEquals
import kotlin.test.assertTrue
/**
* RFC 9000 §8.2 path validation minimum-viable peer-initiated case.
*
* Soak target #4 from the audio-rooms hardening pass: support
* the spec-required PATH_CHALLENGE / PATH_RESPONSE round-trip.
*
* Scope of this landing:
* - Frame codec for PATH_CHALLENGE (0x1A) and PATH_RESPONSE
* (0x1B) was previously decoded-and-discarded.
* - Server-initiated path validation: peer sends
* PATH_CHALLENGE; client MUST echo the SAME 8-byte payload
* in a PATH_RESPONSE on the next outbound packet.
*
* Out of scope (explicit follow-on):
* - Client-initiated migration: requires UdpSocket replacement,
* new-CID acquisition tracking, and validating the new path
* BEFORE moving traffic to it.
* - Anti-amplification on unvalidated paths (RFC 9000 §8.1).
*
* Why this matters even without client-initiated migration: any
* compliant peer (server or middlebox) MAY probe the path at any
* time most commonly after a NAT rebind or our connection-id
* rotation. A peer that doesn't see a PATH_RESPONSE within a few
* RTTs may declare the path dead and tear the connection down,
* visible to users as a sudden audio cut on a phone that briefly
* switched cells.
*/
class PathValidationTest {
@Test
fun pathChallengeFrameRoundTripsThroughCodec() {
val payload = byteArrayOf(0x01, 0x23, 0x45, 0x67, -0x77, -0x55, -0x33, -0x11)
val encoded = encodeFrames(listOf(PathChallengeFrame(payload)))
val decoded = decodeFrames(encoded)
assertEquals(1, decoded.size)
val frame = decoded.first() as PathChallengeFrame
assertContentEquals(
payload,
frame.data,
"PATH_CHALLENGE codec must round-trip the 8-byte payload byte-for-byte",
)
}
@Test
fun pathResponseFrameRoundTripsThroughCodec() {
val payload = byteArrayOf(-0x80, 0x7F, 0x00, -0x01, 0x55, -0x56, 0x42, -0x43)
val encoded = encodeFrames(listOf(PathResponseFrame(payload)))
val decoded = decodeFrames(encoded)
assertEquals(1, decoded.size)
val frame = decoded.first() as PathResponseFrame
assertContentEquals(payload, frame.data)
}
@Test
fun pathChallengeFrameRejectsNonEightByteData() {
try {
PathChallengeFrame(ByteArray(7))
error("PATH_CHALLENGE constructor must reject < 8 bytes")
} catch (_: IllegalArgumentException) {
// expected
}
try {
PathChallengeFrame(ByteArray(9))
error("PATH_CHALLENGE constructor must reject > 8 bytes")
} catch (_: IllegalArgumentException) {
// expected
}
}
@Test
fun inboundPathChallengeQueuesMatchingPathResponse() =
runBlocking {
val (client, pipe) = newConnectedClient()
val challengeData = byteArrayOf(0xCA.toByte(), 0xFE.toByte(), 0xBA.toByte(), 0xBE.toByte(), 0xDE.toByte(), 0xAD.toByte(), 0xBE.toByte(), 0xEF.toByte())
// Server sends a PATH_CHALLENGE in a 1-RTT packet. Pre-fix
// the client would silently absorb it — the connection
// would stay CONNECTED but the peer would never see a
// PATH_RESPONSE, eventually declaring the path dead.
val packet = pipe.buildServerApplicationDatagram(listOf(PathChallengeFrame(challengeData)))!!
feedDatagram(client, packet, nowMillis = 0L)
// Drain the next outbound packet and verify it carries a
// PATH_RESPONSE with the EXACT same 8 bytes.
val outbound = drainOutbound(client, nowMillis = 0L)
assertTrue(outbound != null, "client must emit an outbound packet (ACK + PATH_RESPONSE)")
val frames = pipe.decryptClientApplicationFrames(outbound)
assertTrue(frames != null, "outbound packet must decrypt with the live application keys")
val response = frames.firstOrNull { it is PathResponseFrame } as? PathResponseFrame
assertTrue(response != null, "outbound packet must contain a PATH_RESPONSE — got ${frames.map { it::class.simpleName }}")
assertContentEquals(
challengeData,
response.data,
"PATH_RESPONSE MUST echo the PATH_CHALLENGE payload exactly (byte equality is " +
"the discriminator the peer uses to match a response to its outstanding challenge)",
)
assertEquals(QuicConnection.Status.CONNECTED, client.status)
}
@Test
fun multipleQueuedPathChallengesDrainAsMultipleResponses() =
runBlocking {
// RFC 9000 §8.2: a peer MAY send several PATH_CHALLENGEs
// (e.g. validation retries on packet loss). Each one
// requires its own PATH_RESPONSE — a single response
// doesn't subsume earlier ones because the payload bytes
// are independent random values.
val (client, pipe) = newConnectedClient()
val challenges =
listOf(
ByteArray(8) { 0x10.toByte() },
ByteArray(8) { 0x20.toByte() },
ByteArray(8) { 0x30.toByte() },
)
for (data in challenges) {
val packet = pipe.buildServerApplicationDatagram(listOf(PathChallengeFrame(data)))!!
feedDatagram(client, packet, nowMillis = 0L)
}
// Drain all outbound packets and collect every PATH_RESPONSE
// we see. The writer can fold all three into one packet
// (each is just 9 wire bytes) but might also split — either
// shape is spec-correct.
val responses = mutableListOf<PathResponseFrame>()
while (true) {
val out = drainOutbound(client, nowMillis = 0L) ?: break
val frames = pipe.decryptClientApplicationFrames(out) ?: continue
for (f in frames) {
if (f is PathResponseFrame) responses += f
}
}
assertEquals(
challenges.size,
responses.size,
"client must emit exactly one PATH_RESPONSE per inbound PATH_CHALLENGE",
)
// The responses can arrive in any order; match by content.
val responseSet = responses.map { it.data.toList() }.toSet()
val challengeSet = challenges.map { it.toList() }.toSet()
assertEquals(
challengeSet,
responseSet,
"every challenge payload must appear in some response",
)
}
@Test
fun pathResponseQueueIsBoundedAgainstChallengeFlood() =
runBlocking {
// Defence-in-depth: an attacker spamming PATH_CHALLENGE
// shouldn't pin arbitrary memory in our pendingPathChallengePayloads
// queue. Cap is MAX_PENDING_PATH_RESPONSES (64); excess
// challenges are silently dropped — the protocol allows
// it (peer would retransmit on PTO if a response actually
// mattered).
val (client, pipe) = newConnectedClient()
val flood = QuicConnection.MAX_PENDING_PATH_RESPONSES * 4
for (i in 0 until flood) {
val data = ByteArray(8) { ((i shr 8) and 0xFF).toByte() }
data[7] = (i and 0xFF).toByte()
val packet = pipe.buildServerApplicationDatagram(listOf(PathChallengeFrame(data)))!!
feedDatagram(client, packet, nowMillis = 0L)
}
val responses = mutableListOf<PathResponseFrame>()
while (true) {
val out = drainOutbound(client, nowMillis = 0L) ?: break
val frames = pipe.decryptClientApplicationFrames(out) ?: continue
for (f in frames) {
if (f is PathResponseFrame) responses += f
}
}
assertTrue(
responses.size <= QuicConnection.MAX_PENDING_PATH_RESPONSES,
"response count ${responses.size} must not exceed cap " +
"${QuicConnection.MAX_PENDING_PATH_RESPONSES}",
)
// And: connection survives the flood.
assertEquals(QuicConnection.Status.CONNECTED, client.status)
}
// Default caps — path validation tests don't open streams.
private fun newConnectedClient(): Pair<QuicConnection, InMemoryQuicPipe> =
com.vitorpamplona.quic.connection
.newConnectedClient()
}
@@ -0,0 +1,440 @@
/*
* 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.frame.AckFrame
import com.vitorpamplona.quic.frame.StreamFrame
import com.vitorpamplona.quic.stream.StreamId
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
/**
* Pins the stream-retirement contract that keeps `streamsList` /
* `streams` bounded across long-lived audio-room sessions.
*
* Production motivation: the moq-lite listener path in
* `nestsClient/src/commonMain/kotlin/.../moq/lite/` receives one
* peer-uni stream per Opus frame. At ~50 frames/sec a 3-hour broadcast
* mints ~540 000 streams. Pre-retirement, every stream stayed pinned in
* `streamsList` (insertion-only) and `streams` (no remove) for the
* lifetime of the connection which made `QuicConnection`'s heap grow
* monotonically until the audio room session was torn down. The
* acceptance criterion in the soak prompt is "no monotonic growth past
* handshake-stable"; this file is the unit-test surface that pins the
* mechanics behind that property.
*
* Three angles + a soak-shape:
* 1. [retiredLocalUniStreamsAreRemovedAfterFinAndAck] local-uni send
* path: client opens uni streams, writes + FIN, peer ACKs. Once FIN
* is `markAcked`'d, the next writer drain retires them.
* 2. [retiredPeerInitiatedStreamsDoNotPinReceiveBuffers] moq-lite
* listener path: peer opens server-uni streams, sends payload + FIN,
* parser drains contiguous bytes and closes incoming. Next writer
* drain retires them.
* 3. [retirementPreservesConnectionLevelMaxDataAccounting] the subtle
* correctness check. Removing N retired streams must NOT regress the
* writer's connection-level MAX_DATA accounting; the
* `retiredStreamsRecvBytes` accumulator on the connection folds
* per-stream `receive.contiguousEnd()` sums forward. After retire,
* `advertisedMaxData` continues to grow on subsequent receive bytes.
* 4. [streamChurnSoakKeepsTrackerBounded] soak-shape: cycle through
* 200 generations of 50 streams (10 000 stream lifecycles, well past
* steady-state) and assert the working set stays bounded. Without
* retirement the working set would equal totalStreams.
*/
class StreamRetirementSoakTest {
@Test
fun retiredLocalUniStreamsAreRemovedAfterFinAndAck() =
runBlocking {
val (client, pipe) = newConnectedClient()
val n = 50
val streams =
client.openUniStreamsBatch(items = (0 until n).toList()) { stream, i ->
stream.send.enqueue(
"frame-${i.toString().padStart(2, '0')}".encodeToByteArray(),
)
stream.send.finish()
stream
}
assertEquals(n, client.streamsListLocked().size)
// Drain every outbound packet the writer has to send.
val anyOutput = drainAll(client, pipe)
assertTrue(anyOutput, "writer must have emitted at least one application packet")
// Peer ACKs every PN we ever sent on the application space.
// largestAcked = nextPacketNumber - 1; firstAckRange covers
// PN 0 through largestAcked inclusive.
val largestSent = client.application.pnSpace.nextPacketNumber - 1L
assertTrue(
largestSent >= 0L,
"drainAll must have allocated at least one application PN (got=$largestSent)",
)
val ackPacket =
pipe.buildServerApplicationDatagram(
listOf(
AckFrame(
largestAcknowledged = largestSent,
ackDelay = 0L,
firstAckRange = largestSent,
),
),
)!!
feedDatagram(client, ackPacket, nowMillis = 0L)
for ((idx, s) in streams.withIndex()) {
assertTrue(
s.send.finAcked,
"stream[$idx] (id=${s.streamId}) finAcked must latch after peer ACK",
)
assertTrue(
s.isFullyRetired,
"stream[$idx] uni-out direction is fully retired after finAcked",
)
}
// One more drain triggers retireFullyDoneStreamsLocked at the
// top of buildApplicationPacket — the production seam.
drainAll(client, pipe)
assertEquals(
0,
client.streamsListLocked().size,
"streamsList must be empty after retirement of all FIN-acked uni streams",
)
assertEquals(
n.toLong(),
client.retiredStreamsCount,
"retiredStreamsCount must equal the number of retired streams",
)
assertEquals(
0L,
client.retiredStreamsRecvBytes,
"uni-out streams contribute no receive-side bytes to the accumulator",
)
}
@Test
fun retiredPeerInitiatedStreamsDoNotPinReceiveBuffers() =
runBlocking {
val (client, pipe) = newConnectedClient()
val n = 30
val firstServerUniId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
val serverUniIds = (0 until n).map { firstServerUniId + 4L * it }
val payloads = serverUniIds.associateWith { id -> "g-$id".encodeToByteArray() }
for (id in serverUniIds) {
val frame =
StreamFrame(
streamId = id,
offset = 0L,
data = payloads[id]!!,
fin = true,
)
val packet = pipe.buildServerApplicationDatagram(listOf(frame))!!
feedDatagram(client, packet, nowMillis = 0L)
}
// Every per-stream incoming Flow must complete (parser closed
// it via the isFullyRead branch).
for (id in serverUniIds) {
val stream = client.streamById(id)!!
val chunks = withTimeoutOrNull(2_000L) { stream.incoming.toList() }
assertTrue(chunks != null, "stream $id incoming Flow must terminate after FIN")
val joined = ByteArray(chunks.sumOf { it.size })
var p = 0
for (c in chunks) {
c.copyInto(joined, p)
p += c.size
}
assertEquals(
payloads[id]!!.decodeToString(),
joined.decodeToString(),
"stream $id payload must surface intact",
)
}
// Drain → retireFullyDoneStreamsLocked at the top of
// buildApplicationPacket drops every server-uni stream:
// - direction = UNIDIRECTIONAL_REMOTE_TO_LOCAL → send
// side is trivially settled
// - receive.finReceived AND receive.isFullyRead() because
// the parser drained every chunk into the incoming
// channel before closing it
drainAll(client, pipe)
assertEquals(
0,
client.streamsListLocked().size,
"every server-uni stream must be retired after parsing FIN + writer drain",
)
assertEquals(
n.toLong(),
client.retiredStreamsCount,
"retiredStreamsCount must equal the number of peer-uni streams retired",
)
val expectedRecvBytes = payloads.values.sumOf { it.size }.toLong()
assertEquals(
expectedRecvBytes,
client.retiredStreamsRecvBytes,
"retiredStreamsRecvBytes must aggregate every retired stream's receive high-water",
)
}
@Test
fun phantomGuardDropsRetransmitOnRetiredPeerStream() =
runBlocking {
// Phantom-stream guard regression: after we've retired a
// peer-uni stream, a duplicate STREAM frame the peer
// retransmits (because its loss-detector fired before our
// ACK reached it) MUST NOT mint a fresh QuicStream object.
// Pre-guard, the duplicate frame would:
// - re-populate `streams[id]` and `streamsList`
// - bump `peerInitiatedUniCount` a second time
// - re-fire `newPeerStreams.addLast(...)`, signalling
// a phantom stream-arrival to application code
// - re-deliver the same bytes on a fresh incoming Flow
// The guard at the parser drops the frame (still
// ack-eliciting, so our ACK still goes out and the peer's
// loss-detector quiets down).
val (client, pipe) = newConnectedClient()
val streamId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
val payload = "phantom-test".encodeToByteArray()
val frame = StreamFrame(streamId = streamId, offset = 0L, data = payload, fin = true)
val firstPacket = pipe.buildServerApplicationDatagram(listOf(frame))!!
feedDatagram(client, firstPacket, nowMillis = 0L)
// Drain the original delivery to retire the stream.
client.streamById(streamId)?.incoming?.toList()
drainAll(client, pipe)
assertEquals(
1L,
client.retiredStreamsCount,
"first delivery + drain must retire exactly one stream",
)
assertTrue(
client.isStreamIdRetiredLocked(streamId),
"retired-id ring must hold the just-retired stream id",
)
val countAfterFirstDelivery = client.peerInitiatedUniCount
// Replay the same STREAM+FIN frame — i.e. peer retransmit.
val replayPacket = pipe.buildServerApplicationDatagram(listOf(frame))!!
feedDatagram(client, replayPacket, nowMillis = 0L)
drainAll(client, pipe)
// No phantom stream — counters must not move.
assertEquals(
1L,
client.retiredStreamsCount,
"duplicate STREAM frame must not create a phantom stream that re-retires",
)
assertEquals(
countAfterFirstDelivery,
client.peerInitiatedUniCount,
"peerInitiatedUniCount must not double-count the retransmitted FIN",
)
assertEquals(
0,
client.streamsListLocked().size,
"streamsList must stay empty after the phantom-guard drop",
)
assertEquals(
QuicConnection.Status.CONNECTED,
client.status,
"connection must stay CONNECTED across a duplicate STREAM frame",
)
}
@Test
fun retirementPreservesConnectionLevelMaxDataAccounting() =
runBlocking {
// Without the retiredStreamsRecvBytes seed, the writer's
// `totalRecvAdvanced` would reset to 0 after retirement and
// its half-window MAX_DATA threshold check would never fire
// again — the peer's send credit would silently freeze at
// initialMaxData. Pin the seed by:
// 1. Pushing enough peer bytes to exceed initialMaxData / 2,
// forcing a fresh MAX_DATA frame on the wire.
// 2. Triggering retirement.
// 3. Pushing a second wave that ALSO crosses the half-window
// threshold from the *post-retire* baseline.
// If the writer's seed is wrong, step 3 fails to advance
// advertisedMaxData and MaxDataFrame is not emitted.
val (client, pipe) = newConnectedClient()
val cfg = client.config
val perStream = (cfg.initialMaxStreamDataUni / 4).coerceAtMost(8L * 1024)
val streamCountForFirstWave =
(cfg.initialMaxData / 2 / perStream + 1).toInt().coerceAtLeast(2)
val firstStart = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
for (i in 0 until streamCountForFirstWave) {
val id = firstStart + 4L * i
val payload = ByteArray(perStream.toInt()) { (it and 0xFF).toByte() }
val packet =
pipe.buildServerApplicationDatagram(
listOf(StreamFrame(streamId = id, offset = 0L, data = payload, fin = true)),
)!!
feedDatagram(client, packet, nowMillis = 0L)
client.streamById(id)?.incoming?.toList()
}
// Drain so MAX_DATA goes out + retirement runs.
drainAll(client, pipe)
val advertisedAfterFirstWave = client.advertisedMaxData
assertTrue(
advertisedAfterFirstWave > cfg.initialMaxData,
"first wave must have bumped advertisedMaxData (was=$advertisedAfterFirstWave " +
"initialMaxData=${cfg.initialMaxData}) — fixture pre-condition for the " +
"retirement-preserves-credit test",
)
assertEquals(
0,
client.streamsListLocked().size,
"first wave must be fully retired before the second wave begins",
)
// Second wave — push enough additional bytes that the running
// total (retired + open) crosses the next half-window
// threshold. Without the seed in totalRecvAdvanced, the
// second wave would not raise advertisedMaxData.
val secondStart = firstStart + 4L * streamCountForFirstWave
for (i in 0 until streamCountForFirstWave) {
val id = secondStart + 4L * i
val payload = ByteArray(perStream.toInt()) { ((it + 1) and 0xFF).toByte() }
val packet =
pipe.buildServerApplicationDatagram(
listOf(StreamFrame(streamId = id, offset = 0L, data = payload, fin = true)),
)!!
feedDatagram(client, packet, nowMillis = 0L)
client.streamById(id)?.incoming?.toList()
}
drainAll(client, pipe)
val advertisedAfterSecondWave = client.advertisedMaxData
assertTrue(
advertisedAfterSecondWave > advertisedAfterFirstWave,
"second wave must continue to advance advertisedMaxData past " +
"$advertisedAfterFirstWave (was $advertisedAfterSecondWave) — if it didn't, " +
"retirement regressed the writer's connection-level MAX_DATA accounting",
)
// The cumulative receive total — folded into the writer's
// totalRecvAdvanced via `retiredStreamsRecvBytes` — must
// include every stream from both waves.
val expectedCumulativeRecv = 2L * streamCountForFirstWave * perStream
assertTrue(
client.retiredStreamsRecvBytes >= expectedCumulativeRecv,
"retiredStreamsRecvBytes must include both waves; got=${client.retiredStreamsRecvBytes} " +
"expected≥$expectedCumulativeRecv",
)
}
@Test
fun streamChurnSoakKeepsTrackerBounded() =
runBlocking {
// Soak-shape: simulate the moq-lite listener over many group
// generations. 200 generations × 50 streams = 10 000 stream
// lifecycles. Without retirement, streamsList would hold all
// 10 000 entries. With retirement, the working set stays
// bounded at the per-generation batch.
val (client, pipe) = newConnectedClient()
val perGen = 50
val gens = 200
val totalStreams = perGen * gens
var maxLiveStreams = 0
var nextStreamId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
for (g in 0 until gens) {
for (i in 0 until perGen) {
val payload = ByteArray(8) { ((g * perGen + i) and 0xFF).toByte() }
val packet =
pipe.buildServerApplicationDatagram(
listOf(StreamFrame(streamId = nextStreamId, offset = 0L, data = payload, fin = true)),
)!!
feedDatagram(client, packet, nowMillis = 0L)
client.streamById(nextStreamId)?.incoming?.toList()
nextStreamId += 4L
}
val live = client.streamsListLocked().size
if (live > maxLiveStreams) maxLiveStreams = live
drainAll(client, pipe)
}
assertEquals(
0,
client.streamsListLocked().size,
"after the final retirement pass streamsList must drain fully",
)
assertEquals(
totalStreams.toLong(),
client.retiredStreamsCount,
"retiredStreamsCount must equal every stream the soak ever opened",
)
// Bound: working set should stay near per-generation batch.
// If retirement regressed (e.g. only fired at the end), we
// would observe the full totalStreams here.
assertTrue(
maxLiveStreams <= 2 * perGen,
"tracker working set must stay bounded at ~$perGen but observed $maxLiveStreams " +
"across $gens generations — retirement is leaking",
)
}
/**
* Drain every outbound application packet the writer has to send.
* Returns true if at least one was emitted. The pipe consumes them
* (decrypting each one updates its inbound PN tracking) but does
* not itself reply these tests inject ACKs explicitly.
*/
private fun drainAll(
client: QuicConnection,
pipe: InMemoryQuicPipe,
): Boolean {
var any = false
while (true) {
val out = drainOutbound(client, nowMillis = 0L) ?: break
any = true
// Decrypting bumps the pipe's applicationPnSpace so future
// server-built packets carry an up-to-date `largestAckedInSpace`
// value (avoids the truncated-PN encoder underflow that
// otherwise breaks long sequences).
pipe.decryptClientApplicationFrames(out)
}
return any
}
// moq-lite-shaped fixture: 4 K bidi caps + 64 K peer-uni caps so
// the soak harness can churn 10 K + streams without hitting
// either the peer cap or our advertised cap before the writer's
// periodic MAX_STREAMS_UNI extension fires.
private fun newConnectedClient(): Pair<QuicConnection, InMemoryQuicPipe> =
com.vitorpamplona.quic.connection.newConnectedClient(
maxStreamsBidi = 4096,
maxStreamsUni = 65_536,
maxData = 16L * 1024 * 1024,
)
}
@@ -0,0 +1,337 @@
/*
* 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.tls.PermissiveCertificateValidator
import com.vitorpamplona.quic.transport.UdpSocket
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import java.io.File
import java.net.DatagramSocket
import java.net.InetSocketAddress
import kotlin.test.AfterTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
/**
* Pins the driver close-path cleanup contract for soak target #6:
* "Resource cleanup at session close. Run a session, close it, run
* another, repeat 100×. Verify no thread leaks ([Thread.getAllStackTraces]
* count stable), no socket leaks (netstat count stable), no
* Dispatchers.IO worker buildup."
*
* The driver spawns three things that need clean teardown:
* - A [kotlinx.coroutines.SupervisorJob] parented to the test's scope
* (driver.driverJob).
* - Two child coroutines (read loop + send loop) launched on
* `parentScope.coroutineContext + job + Dispatchers.IO`.
* - A [com.vitorpamplona.quic.transport.UdpSocket] bound to an
* ephemeral OS port.
*
* If any of these leaks across [QuicConnectionDriver.close], a 3-hour
* audio-room session that the user joins+leaves repeatedly (typical
* UX: switch rooms a few times, the OS swaps networks, etc.) eventually
* exhausts file descriptors or blooms the JVM thread pool.
*
* The fixture uses a localhost UDP "blackhole" a DatagramSocket bound
* to an ephemeral port that does NOT speak QUIC. The client driver
* tries to handshake but never succeeds; the test isn't about
* handshake correctness, it's about what happens to the driver's
* resources when the application aborts the session before connect.
*
* Acceptance bands (intentionally generous to avoid CI flakiness):
* - Per-session: connection.status latches CLOSED inside close()'s
* bounded wait, driver.driverJob.isCompleted == true after we
* cancel the parent scope, and a second close() call is a no-op
* (idempotency contract added in round-5 #5).
* - Across 100 sessions: net thread growth < 16, where the typical
* Dispatchers.IO worker pool fluctuation is 48 on this JVM. A
* leak that creates one persistent thread per session would
* produce ~100 net growth; 16 is two orders of magnitude below
* that and well above ambient JVM noise.
*/
class QuicConnectionDriverLifecycleTest {
private val blackhole: DatagramSocket = DatagramSocket(InetSocketAddress("127.0.0.1", 0))
@AfterTest
fun tearDown() {
runCatching { blackhole.close() }
}
@Test
fun closeIsIdempotentAndDriverJobCompletes() =
runBlocking {
val parent = CoroutineScope(SupervisorJob() + Dispatchers.IO)
val (driver, connection, socket) = startDriver(parent)
// Let the driver kick off a few PTO cycles so there's actual
// in-flight crypto when close() runs — exercise the
// close-flush path that actually has work to flush.
delay(50)
driver.close()
// close() launches a teardown coroutine on parentScope and
// returns immediately. Wait for that coroutine to finish.
withTimeoutOrNull(5_000L) { driver.closeTeardownJob?.join() }
?: error("close teardown coroutine never completed within 5 s — driver leaked")
assertEquals(
QuicConnection.Status.CLOSED,
connection.status,
"connection.status must be CLOSED after the close-teardown coroutine joins",
)
assertTrue(
socketIsClosed(socket),
"socket.close() must have run by the time the teardown coroutine completes",
)
// Idempotency: a second close() must be a no-op (round-5 #5
// memoizes the teardown launch). It must not throw, must
// not relaunch teardown, must not produce a new
// closeTeardownJob — the original Job stays.
val originalTeardown = driver.closeTeardownJob
driver.close()
assertTrue(
driver.closeTeardownJob === originalTeardown,
"second close() must reuse the memoized teardown Job (round-5 #5 idempotency)",
)
// Cancel the parent so SupervisorJob children unwind. Then
// assert the driver's job is fully done.
parent.cancel()
withTimeoutOrNull(5_000L) { driver.driverJob.join() }
assertTrue(
driver.driverJob.isCompleted,
"driver.driverJob must be completed after parent.cancel() — observed isCompleted=" +
"${driver.driverJob.isCompleted} active=${driver.driverJob.isActive}",
)
}
@Test
fun repeatedSessionLifecycleDoesNotLeakThreads() =
runBlocking {
// Warm up Dispatchers.IO so its worker count has stabilised
// before we sample. Otherwise the first session's growth
// (creating fresh IO workers from the pool's lazy init)
// shows up as a leak.
warmUpDispatchersIo()
val baseline = liveThreadCount()
val baselineFds = liveFileDescriptorCount()
val sessions = 100
for (i in 0 until sessions) {
val parent = CoroutineScope(SupervisorJob() + Dispatchers.IO)
val (driver, connection, socket) = startDriver(parent)
// Keep the session very short — just long enough that
// the driver kicked off both read + send loops and may
// have done a PTO retransmit.
delay(10)
driver.close()
withTimeoutOrNull(2_000L) { driver.closeTeardownJob?.join() }
?: error(
"session $i: close teardown coroutine never completed — driver leaked. " +
"Status=${connection.status}",
)
parent.cancel()
withTimeoutOrNull(2_000L) { driver.driverJob.join() }
?: error("session $i: driver job did not complete after parent.cancel()")
assertEquals(
QuicConnection.Status.CLOSED,
connection.status,
"session $i: connection must be CLOSED after teardown",
)
assertTrue(
socketIsClosed(socket),
"session $i: UDP socket must be closed",
)
}
// Coroutines may still be wrapping up; give Dispatchers.IO
// a small drain window so any in-flight worker reuse
// settles before we sample.
delay(200)
val finalCount = liveThreadCount()
val growth = finalCount - baseline
// Generous band — the IO pool naturally fluctuates a few
// workers. A real leak (one persistent thread per session)
// would push growth ≥ ~100. 16 is strictly above the
// observed JVM noise band and well below leak.
assertTrue(
growth <= 16,
"thread count grew by $growth across $sessions sessions " +
"(baseline=$baseline final=$finalCount). Anything > 16 indicates a leak.",
)
// FD-leak canary. On Linux, /proc/self/fd holds one entry per
// open file descriptor (sockets + pipes + regular files).
// A leak that misses socket.close() would show up here as
// ~1 FD per session — 100 sessions → growth ≥ 100 with
// certainty. We band at 16 (same headroom as threads). On
// platforms without /proc, [liveFileDescriptorCount] returns
// -1 and this branch silently no-ops.
val finalFds = liveFileDescriptorCount()
if (baselineFds >= 0 && finalFds >= 0) {
val fdGrowth = finalFds - baselineFds
assertTrue(
fdGrowth <= 16,
"FD count grew by $fdGrowth across $sessions sessions " +
"(baseline=$baselineFds final=$finalFds). UDP socket / pipe leak.",
)
}
}
@Test
fun socketDeathMidSessionFlipsConnectionToClosed() =
runBlocking {
// Soak target #3: when Android backgrounds the app and the
// OS reclaims the UDP socket FD ~30 s later, the next
// `socket.send` from the QUIC driver throws. Without the
// try/catch on the send loop, that throw escapes silently
// into the SupervisorJob and the connection sits in
// HANDSHAKING / CONNECTED forever — the
// ReconnectingNestsListener's terminal-state listener
// never fires, the room screen shows "live" while audio
// is dead, and the user has to back out and rejoin.
//
// Pin the contract: closing the socket out from under a
// running driver MUST flip connection.status to CLOSED
// within a bounded window (the next send-loop iteration —
// typically the next PTO firing). The reconnect orchestrator
// observes this as terminal and reschedules a fresh handshake.
val parent = CoroutineScope(SupervisorJob() + Dispatchers.IO)
val (driver, connection, socket) = startDriver(parent)
// Let the driver get past start() and into the read+send
// loop steady state — it doesn't matter that the handshake
// isn't completing (no peer); the send loop is alive and
// will hit the dead socket on its next drain or PTO.
delay(50)
// The driver's status here is HANDSHAKING (no peer to
// complete it). Verify that's our pre-condition rather
// than CLOSED — otherwise the test would trivially pass.
assertTrue(
connection.status != QuicConnection.Status.CLOSED,
"pre-condition: connection must NOT yet be CLOSED (was ${connection.status})",
)
// Simulate the OS killing the socket. UdpSocket.close()
// is the same path Android takes when the kernel reclaims
// a backgrounded app's FDs.
socket.close()
// Wait for the send loop to actually attempt a send and
// raise. The send loop fires every PTO; the first PTO is
// ~1 s for a connection without an RTT sample. Give it 3 s
// of headroom.
val startedAt = System.nanoTime()
while (connection.status != QuicConnection.Status.CLOSED &&
(System.nanoTime() - startedAt) < 5_000_000_000L // 5 s in nanos
) {
delay(50)
}
assertEquals(
QuicConnection.Status.CLOSED,
connection.status,
"connection must transition to CLOSED after socket dies; pre-fix it would " +
"stay HANDSHAKING/CONNECTED indefinitely because the send-loop throw " +
"escaped the SupervisorJob silently",
)
// closeReason should be populated for observability — the
// ReconnectingNestsListener orchestrator surfaces this in
// its NestsListenerState.Failed.reason. Either the read
// loop's finally (socket.receive returns null on close) or
// the send loop's catch (socket.send throws on closed
// socket) gets there first depending on scheduler timing;
// both produce a human-readable message that mentions the
// loop. The exact path is racy, so we just check both
// possible reason strings cover the symptom.
val reason = connection.closeReason
assertTrue(
reason != null && (reason.contains("read loop") || reason.contains("send loop")),
"closeReason must mention the loop death so observability surfaces the cause; " +
"got '$reason'",
)
// Cleanup — the driver's job tree should still wind down
// cleanly even though we tore the socket out from under it.
driver.close()
withTimeoutOrNull(2_000L) { driver.closeTeardownJob?.join() }
parent.cancel()
withTimeoutOrNull(2_000L) { driver.driverJob.join() }
assertTrue(
driver.driverJob.isCompleted,
"driver job must complete cleanly even after a socket-death teardown",
)
}
/** True if a UDP send on the socket throws — i.e. close() has run. */
private fun socketIsClosed(socket: UdpSocket): Boolean =
try {
runBlocking { socket.send(ByteArray(1)) }
false
} catch (_: Throwable) {
true
}
private suspend fun startDriver(parent: CoroutineScope): Triple<QuicConnectionDriver, QuicConnection, UdpSocket> {
val socket = UdpSocket.connect("127.0.0.1", blackhole.localPort)
val connection =
QuicConnection(
serverName = "lifecycle.test",
config = QuicConnectionConfig(),
tlsCertificateValidator = PermissiveCertificateValidator(),
)
val driver = QuicConnectionDriver(connection, socket, parent)
driver.start()
return Triple(driver, connection, socket)
}
private suspend fun warmUpDispatchersIo() {
// Touch Dispatchers.IO from a few coroutines so the lazy worker
// pool has its baseline workers spun up. Without this the
// first measurement's "baseline" is artificially low and the
// first sessions' creation of IO workers looks like a leak.
repeat(4) {
kotlinx.coroutines.withContext(Dispatchers.IO) { Thread.sleep(2) }
}
}
private fun liveThreadCount(): Int = Thread.getAllStackTraces().keys.size
/**
* Linux: count entries in `/proc/self/fd`. macOS / Windows / other
* platforms don't expose this path; return -1 so the caller
* silently skips the FD assertion. This is a strictly additive
* canary no false-positive risk on platforms that can't measure.
*/
private fun liveFileDescriptorCount(): Int =
try {
val procFd = File("/proc/self/fd")
if (procFd.isDirectory) procFd.list()?.size ?: -1 else -1
} catch (_: Throwable) {
-1
}
}
@@ -0,0 +1,225 @@
/*
* 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.frame.StreamFrame
import com.vitorpamplona.quic.stream.StreamId
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
/**
* Long-form heap-sampling canary for soak target #1 (memory growth).
*
* **DEFAULT-SKIPPED.** This test is gated by the `quicSoakSeconds`
* system property, propagated from the `-PquicSoakSeconds=N` Gradle
* property by `quic/build.gradle.kts`. Without the property the
* test method early-returns with a printed "SKIPPED" line, so a
* plain `./gradlew test` run keeps it out of CI's critical path
* (~40 ms overhead from the runBlocking + property check).
*
* To run the production-shaped 30-minute soak from the audio-rooms
* prompt:
*
* ./gradlew :quic:jvmTest \
* --tests 'com.vitorpamplona.quic.connection.QuicHeapSoakTest' \
* -PquicSoakSeconds=1800
*
* Quick local sanity (30 seconds, ~25 K stream lifecycles):
*
* ./gradlew :quic:jvmTest \
* --tests 'com.vitorpamplona.quic.connection.QuicHeapSoakTest' \
* -PquicSoakSeconds=30
*
* Test shape: drive moq-lite-shaped peer-uni stream churn through an
* [InMemoryQuicPipe] at a fixed rate. Sample
* `Runtime.getRuntime().totalMemory() - freeMemory()` at six evenly-
* spaced points across the run, calling `System.gc()` first to
* minimise allocator-noise. The post-warmup baseline is the second
* sample; the acceptance criterion (10 MB max growth past warmup,
* direct from the soak prompt) is checked against the final sample.
*
* Why six samples: enough granularity to spot a slow drift while
* keeping the per-sample overhead small relative to the run. The
* second sample (~17 % into the run) is the warmup baseline the
* first sample is just before any work, so it understates ambient
* usage and would inflate apparent growth.
*/
class QuicHeapSoakTest {
@Test
fun heapStaysFlatUnderModulatedStreamChurn() =
runBlocking {
val durationSec =
System.getProperty("quicSoakSeconds")?.toIntOrNull()
if (durationSec == null || durationSec <= 0) {
// Default-skip path: `./gradlew test` runs this test as a
// no-op fast-pass. Opt in via `-PquicSoakSeconds=N` (1800
// for the 30-minute soak from the audio-rooms prompt).
// We don't use kotlin.test's assumption because the JVM
// backend differs between JUnit4 / JUnit5; an early
// return prints the reason and keeps CI fast.
println(
"[QuicHeapSoakTest] SKIPPED — set -PquicSoakSeconds=N " +
"to run (e.g. 30 for sanity, 1800 for production-shape).",
)
return@runBlocking
}
// Open the pipe; the handshake should be cheap relative to
// the soak duration.
val (client, pipe) = newConnectedClient()
// moq-lite production rate: ~50 peer-uni streams per second
// (one Opus frame per stream). A 1 800 s run mints 90 000
// streams; a 30 s scaled-down run still moves through 1 500
// generations of churn — plenty for retirement to fire
// hundreds of times.
val streamsPerSecond = 50
val totalStreams = durationSec.toLong() * streamsPerSecond
val perPayload = 32 // ~Opus frame envelope
val sampleCount = 6
val streamsPerSample = (totalStreams / sampleCount).coerceAtLeast(1L)
val samples = LongArray(sampleCount)
val streamCountAtSample = LongArray(sampleCount)
var nextStreamId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
var streamsDelivered = 0L
var sampleIdx = 0
while (sampleIdx < sampleCount) {
val target = (sampleIdx + 1) * streamsPerSample
while (streamsDelivered < target) {
val payload = ByteArray(perPayload) { (streamsDelivered.toInt() and 0xFF).toByte() }
val packet =
pipe.buildServerApplicationDatagram(
listOf(
StreamFrame(
streamId = nextStreamId,
offset = 0L,
data = payload,
fin = true,
),
),
)!!
feedDatagram(client, packet, nowMillis = 0L)
client.streamById(nextStreamId)?.incoming?.toList()
nextStreamId += 4L
streamsDelivered += 1
// Periodically drain so retireFullyDoneStreamsLocked
// runs at moq-lite-realistic intervals (per ~50
// streams). Without periodic drains the working set
// grows to streamsPerSample which is misleading vs
// the production rate.
if (streamsDelivered % streamsPerSecond == 0L) {
drainAll(client, pipe)
}
}
drainAll(client, pipe)
samples[sampleIdx] = sampleHeapBytes()
streamCountAtSample[sampleIdx] = client.streamsListLocked().size.toLong()
sampleIdx += 1
}
// Sanity: every stream we *actually* minted (streamsDelivered;
// streamsPerSample integer-truncates so it can be < totalStreams)
// must have been retired by the last sample. drainAll between
// batches drives retirement. If retirement regressed, the gap
// here would surface immediately.
assertEquals(
streamsDelivered,
client.retiredStreamsCount,
"every minted peer-uni stream must have been retired by end of soak",
)
// Working set must have stayed bounded throughout — never
// larger than the per-sample batch size (drainAll runs at
// every streamsPerSecond boundary, so the steady-state
// bound is ≤ streamsPerSecond entries).
val maxWorkingSet = streamCountAtSample.max()
assertTrue(
maxWorkingSet <= streamsPerSecond.toLong(),
"tracker working set must stay ≤ $streamsPerSecond entries; observed max=$maxWorkingSet " +
"(samples=${streamCountAtSample.toList()})",
)
// Heap acceptance — direct from the audio-rooms prompt:
// "no monotonic growth past handshake-stable (10 MB)".
// Use sample 1 (post-warmup) as baseline, sample 5 (final)
// as the steady-state probe.
val warmup = samples[1]
val finalSample = samples[sampleCount - 1]
val growthMb = (finalSample - warmup).toDouble() / (1024.0 * 1024.0)
// Print the sample profile so a CI failure has actionable
// forensic data (which sample first crossed the threshold,
// roughly when in the run).
val profile =
samples.indices.joinToString(prefix = "[", postfix = "]") { i ->
val mb = samples[i].toDouble() / (1024.0 * 1024.0)
"%.1fMB".format(mb)
}
System.err.println(
"[QuicHeapSoakTest] duration=${durationSec}s totalStreams=$totalStreams " +
"samples=$profile retired=${client.retiredStreamsCount}",
)
assertTrue(
growthMb <= 10.0,
"heap grew by %.1f MB past warmup (sample 1 = %.1f MB → final = %.1f MB); ".format(
growthMb,
warmup.toDouble() / 1024.0 / 1024.0,
finalSample.toDouble() / 1024.0 / 1024.0,
) + "samples=$profile. Acceptance: ≤ 10 MB.",
)
}
private fun sampleHeapBytes(): Long {
// Best-effort GC: a single System.gc() is a hint, not a
// guarantee. Three passes with a small sleep between gives
// G1/CMS/ZGC enough time to actually settle. (We avoid
// `System.runFinalization()` — deprecated since Java 18 and
// unreliable on modern collectors.)
repeat(3) {
System.gc()
Thread.sleep(20)
}
val rt = Runtime.getRuntime()
return rt.totalMemory() - rt.freeMemory()
}
private fun drainAll(
client: QuicConnection,
pipe: InMemoryQuicPipe,
) {
while (true) {
val out = drainOutbound(client, nowMillis = 0L) ?: break
pipe.decryptClientApplicationFrames(out)
}
}
// moq-lite-shaped fixture (matches StreamRetirementSoakTest's
// shape) — large peer-uni cap so the heap canary can churn 90 K
// streams in a 30-min run without bumping the cap.
private fun newConnectedClient(): Pair<QuicConnection, InMemoryQuicPipe> =
com.vitorpamplona.quic.connection.newConnectedClient(
maxStreamsBidi = 4096,
maxStreamsUni = 65_536,
maxData = 16L * 1024 * 1024,
)
}