diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt index 99703e822..4ac9fefc0 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt @@ -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 diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/nests/AppForegroundRecycleHook.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/nests/AppForegroundRecycleHook.kt new file mode 100644 index 000000000..cc86f5fee --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/nests/AppForegroundRecycleHook.kt @@ -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 + } +} diff --git a/amethyst/src/test/java/com/vitorpamplona/amethyst/service/nests/AppForegroundRecycleHookTest.kt b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/nests/AppForegroundRecycleHookTest.kt new file mode 100644 index 000000000..11e53daba --- /dev/null +++ b/amethyst/src/test/java/com/vitorpamplona/amethyst/service/nests/AppForegroundRecycleHookTest.kt @@ -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) + } +} diff --git a/quic/build.gradle.kts b/quic/build.gradle.kts index 83503de9b..def6dad69 100644 --- a/quic/build.gradle.kts +++ b/quic/build.gradle.kts @@ -124,3 +124,14 @@ tasks.register("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().configureEach { + val soakSeconds = (project.findProperty("quicSoakSeconds") as? String) + if (soakSeconds != null) { + systemProperty("quicSoakSeconds", soakSeconds) + } +} diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index 0228ae728..4682d7f6a 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -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() 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() + private val retiredStreamIdSet = HashSet() + /** * Peer-advertised concurrent bidirectional stream cap. Initialised from * [TransportParameters.initialMaxStreamsBidi] when peer params arrive, @@ -413,6 +480,31 @@ class QuicConnection( internal val pendingNewConnectionId: MutableMap = 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 = 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 = 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 = 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 } } diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt index 4c8fab2ec..75d8b2e25 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -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 diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt index 1ac2c1a0a..738884406 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -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 diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 607279620..3172eb52b 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -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() // 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 diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/frame/Frame.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/frame/Frame.kt index d61fd774f..965b0e529 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/frame/Frame.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/frame/Frame.kt @@ -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 { } 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 -> { diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt index 05317cb3b..46d1cd888 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -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- diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CloseUnderLoadTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CloseUnderLoadTest.kt new file mode 100644 index 000000000..831685c71 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CloseUnderLoadTest.kt @@ -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.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 = + 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 } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ConnectedClientFixture.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ConnectedClientFixture.kt new file mode 100644 index 000000000..c009322bf --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/ConnectedClientFixture.kt @@ -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 = + 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 + } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt index 9957825b1..d1b21ad0f 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/InMemoryQuicPipe.kt @@ -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): 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, diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/KeyUpdatePeerInitiatedTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/KeyUpdatePeerInitiatedTest.kt new file mode 100644 index 000000000..09d444392 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/KeyUpdatePeerInitiatedTest.kt @@ -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 = + com.vitorpamplona.quic.connection + .newConnectedClient() +} diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MoqLiteLossHarnessTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MoqLiteLossHarnessTest.kt new file mode 100644 index 000000000..e818f4915 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MoqLiteLossHarnessTest.kt @@ -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() + + 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() + 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() + + 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() + 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 = + com.vitorpamplona.quic.connection.newConnectedClient( + maxStreamsBidi = 1024, + maxStreamsUni = 1024, + maxData = 16L * 1024 * 1024, + ) +} diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PathValidationTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PathValidationTest.kt new file mode 100644 index 000000000..aba59f40c --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PathValidationTest.kt @@ -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() + 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() + 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 = + com.vitorpamplona.quic.connection + .newConnectedClient() +} diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetirementSoakTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetirementSoakTest.kt new file mode 100644 index 000000000..df8b51400 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetirementSoakTest.kt @@ -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 = + com.vitorpamplona.quic.connection.newConnectedClient( + maxStreamsBidi = 4096, + maxStreamsUni = 65_536, + maxData = 16L * 1024 * 1024, + ) +} diff --git a/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriverLifecycleTest.kt b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriverLifecycleTest.kt new file mode 100644 index 000000000..8631a50ac --- /dev/null +++ b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriverLifecycleTest.kt @@ -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 4–8 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 { + 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 + } +} diff --git a/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicHeapSoakTest.kt b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicHeapSoakTest.kt new file mode 100644 index 000000000..97c34292c --- /dev/null +++ b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicHeapSoakTest.kt @@ -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 = + com.vitorpamplona.quic.connection.newConnectedClient( + maxStreamsBidi = 4096, + maxStreamsUni = 65_536, + maxData = 16L * 1024 * 1024, + ) +}