From be8f2e08d311797d89465f97472d4161f0262934 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 23:20:27 +0000 Subject: [PATCH] feat(quic+amethyst): close-under-load + key-update PN gate + loss harness depth + foreground recycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Working through the punch list from the prior "what's left?" status. #2 close-under-load (CloseUnderLoadTest, 3 cases) ================================================== Pins three races between connection close and active stream traffic that the existing idle-driver close test doesn't cover: - closeWhileBulkStreamRetirementIsRunning — server ACKs 100 in-flight client-bidi streams in one shot, retire pass + close fire concurrently. Asserts CLOSED status and no Flow leak. - closeWhileAppCoroutinesAreOpeningStreamsDoesNotDeadlock — pins the lock-ordering invariant: streamsLock (openers) and lifecycleLock (close) don't fight. - closeWhilePeerStreamsAreInFlight — close fires mid-stream of 50 server-uni group streams (half FIN'd, half not). Every incoming Flow terminates promptly with whatever bytes the parser had already delivered. #3 PN-gate / try-previous-fall-through-to-next for key updates ============================================================== Closes the KNOWN-LIMITATION I documented in the prior round. QuicConnectionParser previously routed mismatched-KEY_PHASE packets unconditionally to previousReceiveProtection if non-null, which silently dropped consecutive-rotation packets (KEY_PHASE wraps back to its prior value, prior keys are now wrong, AEAD fails, connection wedges). Fix follows neqo's shape: try previous keys; on AEAD failure fall through to next-phase derivation. Two AEAD attempts on a mismatched-phase packet are cheap; KEY_PHASE mismatch is rare. The previously-disabled twoConsecutiveRotationsCommitCorrectly test now passes. #4 loss harness depth (MoqLiteLossHarnessTest, 3 added cases) ============================================================= First-pass harness from the previous round was a single 5%-loss moq-lite shape. Added: - listenerToleratesPacketReorderingOnGroupStreams — random permutation of 50 group-stream datagrams, asserts 100% delivery. Pins the reorder contract for moq-lite. - listenerSurvivesExtremeTwentyPercentLoss — 200 streams at 20% loss, asserts ≥ 60% delivery and connection stays CONNECTED. Catches catastrophic-collapse regressions in flow-control / ACK-tracker / retired-id ring under stress. - reliableBidiStreamRecoversFromMidStreamPacketLoss — drops the middle two of four STREAM frames on a reliable bidi stream, retransmits, asserts the consumer surfaces the full contiguous payload. Pins the reliability contract distinct from the best-effort moq-lite path. #1 foreground-resume recycle (AppForegroundRecycleHook, 5 tests) ================================================================ Closes the user-visible production gap. ReconnectingNestsListener already orchestrates retry on terminal state and observes NestNetworkChangeBus for network-handover recycles. The missing piece was a foregrounding signal: when Android reclaims the app's UDP socket FD after backgrounding (typical at ~30 s+, network itself still up so the connectivity callback doesn't fire), the QUIC connection sits dead until the next send-loop throw — which landed last round. This hook publishes a NestNetworkChangeBus event when the app returns to foreground after spending ≥ 5 s in background. The pre-existing wiring observes that event and calls recycleSession() on every active listener / speaker. Pure-state core (AppForegroundCounter) is testable without Robolectric; JUnit-4 unit tests pin the threshold logic, multi-activity counter behaviour (e.g. PIP), and consecutive-cycle correctness. Wired into Amethyst.Application.onCreate via registerActivityLifecycleCallbacks. https://claude.ai/code/session_018KPKWRg5baX5Anf7zfEyec --- .../com/vitorpamplona/amethyst/Amethyst.kt | 12 + .../service/nests/AppForegroundRecycleHook.kt | 173 +++++++++ .../nests/AppForegroundRecycleHookTest.kt | 180 +++++++++ .../quic/connection/QuicConnectionParser.kt | 111 ++++-- .../quic/connection/CloseUnderLoadTest.kt | 353 ++++++++++++++++++ .../connection/KeyUpdatePeerInitiatedTest.kt | 62 +-- .../quic/connection/MoqLiteLossHarnessTest.kt | 208 +++++++++++ 7 files changed, 1044 insertions(+), 55 deletions(-) create mode 100644 amethyst/src/main/java/com/vitorpamplona/amethyst/service/nests/AppForegroundRecycleHook.kt create mode 100644 amethyst/src/test/java/com/vitorpamplona/amethyst/service/nests/AppForegroundRecycleHookTest.kt create mode 100644 quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CloseUnderLoadTest.kt 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/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt index 87922b550..be1632c3e 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -301,26 +301,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 +378,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", 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..c5702ae69 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/CloseUnderLoadTest.kt @@ -0,0 +1,353 @@ +/* + * 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 com.vitorpamplona.quic.tls.InProcessTlsServer +import com.vitorpamplona.quic.tls.PermissiveCertificateValidator +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) + } + } + + private fun newConnectedClient(): Pair = + runBlocking { + val client = + QuicConnection( + serverName = "closeunderload.test", + config = + QuicConnectionConfig( + initialMaxStreamsBidi = 1024, + initialMaxStreamsUni = 1024, + initialMaxData = 16L * 1024 * 1024, + initialMaxStreamDataBidiLocal = 64L * 1024, + initialMaxStreamDataBidiRemote = 64L * 1024, + initialMaxStreamDataUni = 64L * 1024, + ), + tlsCertificateValidator = PermissiveCertificateValidator(), + ) + val serverScid = ConnectionId.random(8) + val tlsServer = + InProcessTlsServer( + transportParameters = + TransportParameters( + initialMaxData = 16L * 1024 * 1024, + initialMaxStreamDataBidiLocal = 64L * 1024, + initialMaxStreamDataBidiRemote = 64L * 1024, + initialMaxStreamDataUni = 64L * 1024, + initialMaxStreamsBidi = 1024, + initialMaxStreamsUni = 1024, + 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 = 16) + assertEquals(QuicConnection.Status.CONNECTED, client.status) + client to pipe + } +} + +/** + * 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/KeyUpdatePeerInitiatedTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/KeyUpdatePeerInitiatedTest.kt index 5dedf506e..715e2f838 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/KeyUpdatePeerInitiatedTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/KeyUpdatePeerInitiatedTest.kt @@ -265,29 +265,45 @@ class KeyUpdatePeerInitiatedTest { assertEquals(QuicConnection.Status.CONNECTED, client.status) } - // KNOWN LIMITATION — consecutive rotations within the reorder window. - // - // The parser currently routes inbound packets whose KEY_PHASE bit - // does not match `currentReceiveKeyPhase` to - // `previousReceiveProtection` whenever those prior keys exist. After - // a single rotation that's correct (matches RFC 9001 §6.1's reorder- - // tolerance contract). But after a SECOND rotation by the peer the - // KEY_PHASE bit wraps back to the original value — and our parser - // misroutes those packets to the (now-wrong) prior keys, AEAD-fails, - // and silently drops them. The connection wedges. - // - // The standard fix is to gate `previousReceiveProtection` on a - // packet-number threshold (track the PN of the rotation-triggering - // packet; reroute to next-phase derivation once subsequent KEY_PHASE- - // mismatched packets exceed it). neqo and picoquic implement this; - // we don't yet. For audio-rooms' 3-hour session, ONE rotation is the - // realistic case (peers rotate sparingly), so the deeper fix is a - // follow-up rather than a blocker. - // - // No test asserts the broken behaviour — adding one would just pin - // the regression. When the protocol-level fix lands, a positive - // `consecutiveRotationsCommitCorrectly` test is the right addition - // here. + @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 diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MoqLiteLossHarnessTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MoqLiteLossHarnessTest.kt index c17282b53..b6180277f 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MoqLiteLossHarnessTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/MoqLiteLossHarnessTest.kt @@ -30,6 +30,7 @@ import kotlinx.coroutines.withTimeoutOrNull import kotlin.random.Random import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertNotNull import kotlin.test.assertTrue /** @@ -184,6 +185,213 @@ class MoqLiteLossHarnessTest { 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) + } + private fun newConnectedClient(): Pair = runBlocking { val client =