feat(quic+amethyst): close-under-load + key-update PN gate + loss harness depth + foreground recycle

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
2026-05-07 23:20:27 +00:00
parent 6706c111b5
commit be8f2e08d3
7 changed files with 1044 additions and 55 deletions
@@ -22,6 +22,7 @@ package com.vitorpamplona.amethyst
 import android.app.Application
 import com.vitorpamplona.amethyst.service.logging.Logging
 import com.vitorpamplona.amethyst.service.nests.AppForegroundRecycleHook
 import com.vitorpamplona.quartz.utils.Log
 import com.vitorpamplona.quartz.utils.LogLevel
@@ -41,6 +42,17 @@ class Amethyst : Application() {
        Log.d("AmethystApp") { "onCreate $this" }
        instance = AppModules(this)
        // After-background foreground recycle: when the app returns to
        // the foreground after spending more than ~5 s in the
        // background, publish a network-change event so every active
        // NestViewModel recycles its underlying QUIC session. Covers
        // the case where Android reclaims our UDP socket FD while
        // backgrounded — the connectivity callback in
        // `NestForegroundService` doesn't fire there because the
        // network itself is still up. See `AppForegroundRecycleHook`'s
        // kdoc for the threshold rationale.
        registerActivityLifecycleCallbacks(AppForegroundRecycleHook())
        if (isDebug) {
            Logging.setup()
            // Auto-enable the Nests session-trace recorder in debug
@@ -0,0 +1,173 @@
 /*
 * Copyright (c) 2025 Vitor Pamplona
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to use,
 * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
 * Software, and to permit persons to whom the Software is furnished to do so,
 * subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
 * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
 * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
 * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
 * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */
 package com.vitorpamplona.amethyst.service.nests
 import android.app.Activity
 import android.app.Application
 import android.os.Bundle
 import com.vitorpamplona.amethyst.commons.viewmodels.NestNetworkChangeBus
 import com.vitorpamplona.quartz.utils.Log
 /**
 * Pure-state foreground/background tracker, decoupled from
 * `android.app.Activity` so the unit tests can drive it without
 * Robolectric or Mockito.
 *
 * See [AppForegroundRecycleHook] for the production motivation /
 * threshold-rationale kdoc — this class is just the testable core.
 *
 * Threading: all state-mutating methods are documented to run on the
 * Android main thread (Application lifecycle callbacks fire there);
 * tests call them serially, so no synchronisation is needed.
 */
 class AppForegroundCounter(
    private val backgroundThresholdMs: Long = AppForegroundRecycleHook.DEFAULT_BACKGROUND_THRESHOLD_MS,
    private val publishEvent: () -> Unit = { NestNetworkChangeBus.publish() },
    private val nowMillis: () -> Long = { System.currentTimeMillis() },
 ) {
    private var startedActivities = 0
    private var lastBackgroundedAtMillis: Long = -1L
    /**
     * Count of recycle events fired since construction. Diagnostic
     * surface for tests; production code observes the side-effect via
     * [NestNetworkChangeBus] instead.
     */
    var recyclesFired: Int = 0
        private set
    /**
     * Increment the started-activity counter and, if this is the
     * 0 → 1 transition AND the app spent ≥ [backgroundThresholdMs]
     * in the background, fire [publishEvent]. The first
     * onActivityStarted after process start is a no-op (no prior
     * background timestamp to compare against).
     */
    fun onActivityStarted() {
        val wasBackgrounded = startedActivities == 0
        startedActivities++
        if (!wasBackgrounded) return
        val backgroundedAt = lastBackgroundedAtMillis
        if (backgroundedAt < 0L) return
        val backgroundedFor = nowMillis() - backgroundedAt
        if (backgroundedFor < backgroundThresholdMs) {
            Log.d("AppForegroundCounter") {
                "skipping recycle on resume after only ${backgroundedFor}ms background " +
                    "(threshold=${backgroundThresholdMs}ms)"
            }
            return
        }
        Log.d("AppForegroundCounter") {
            "publishing recycle event on resume after ${backgroundedFor}ms background"
        }
        recyclesFired++
        publishEvent()
    }
    /**
     * Decrement the counter; on N → 0 transition, record the
     * background timestamp.
     */
    fun onActivityStopped() {
        startedActivities--
        if (startedActivities <= 0) {
            startedActivities = 0
            lastBackgroundedAtMillis = nowMillis()
        }
    }
 }
 /**
 * Application-wide observer that publishes a
 * [NestNetworkChangeBus] event when the app returns to the foreground
 * after spending more than [backgroundThresholdMs] in the background.
 *
 * Production motivation: Android may reclaim a backgrounded app's
 * UDP-socket file descriptors as it ages out of the foreground app
 * pool (the kernel's watcher trims after roughly 30 s of no foreground
 * activity, with quite a bit of variance per OEM). When the user
 * resumes the app, the QUIC connection sitting on the now-reclaimed
 * socket has dead OS-level state, but the connection-level FSM
 * doesn't know that yet — the next `socket.send` throws, and only
 * then does the send-loop catch surface CLOSED.
 *
 * The downstream
 * [com.vitorpamplona.amethyst.commons.viewmodels.NestViewModel]
 * already observes [NestNetworkChangeBus] for the network-handover
 * case (Wi-Fi ↔ cellular). We piggy-back on the same bus here: a
 * "long enough background" event has the same shape from the QUIC
 * driver's perspective as a network change — recycle the underlying
 * session and let the [com.vitorpamplona.nestsclient.connectReconnectingNestsListener]
 * / `connectReconnectingNestsSpeaker` orchestrators reconnect.
 *
 * Why a threshold instead of fire-on-every-resume:
 *  - A short background (notification pulldown, biometric auth, lock-
 *    screen glance) lasts < 1 s and the socket is still healthy. A
 *    forced recycle there is a wasted ~1 s re-handshake gap of audio
 *    silence — annoying to users for no benefit.
 *  - A long background (call from another app, screen off > 30 s,
 *    home-button-then-back) commonly leaves the socket dead. Better
 *    to eat one re-handshake gap than 30 s of silence while the QUIC
 *    PTO times out.
 * 5 seconds is the sweet spot: well over typical UI transitions but
 * well under any plausible socket-reclaim window.
 */
 class AppForegroundRecycleHook(
    backgroundThresholdMs: Long = DEFAULT_BACKGROUND_THRESHOLD_MS,
    publishEvent: () -> Unit = { NestNetworkChangeBus.publish() },
    nowMillis: () -> Long = { System.currentTimeMillis() },
 ) : Application.ActivityLifecycleCallbacks {
    private val counter = AppForegroundCounter(backgroundThresholdMs, publishEvent, nowMillis)
    override fun onActivityStarted(activity: Activity) {
        counter.onActivityStarted()
    }
    override fun onActivityStopped(activity: Activity) {
        counter.onActivityStopped()
    }
    override fun onActivityCreated(
        activity: Activity,
        savedInstanceState: Bundle?,
    ) = Unit
    override fun onActivityResumed(activity: Activity) = Unit
    override fun onActivityPaused(activity: Activity) = Unit
    override fun onActivitySaveInstanceState(
        activity: Activity,
        outState: Bundle,
    ) = Unit
    override fun onActivityDestroyed(activity: Activity) = Unit
    companion object {
        /**
         * Default 5 000 ms — well above the longest plausible UI
         * transition (notification pull, biometric prompt) and well
         * below the 30 s timing the Android kernel uses to reclaim
         * idle UDP sockets.
         */
        const val DEFAULT_BACKGROUND_THRESHOLD_MS = 5_000L
    }
 }
@@ -0,0 +1,180 @@
 /*
 * Copyright (c) 2025 Vitor Pamplona
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to use,
 * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
 * Software, and to permit persons to whom the Software is furnished to do so,
 * subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
 * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
 * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
 * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
 * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */
 package com.vitorpamplona.amethyst.service.nests
 import org.junit.Assert.assertEquals
 import org.junit.Test
 /**
 * Unit tests for [AppForegroundCounter] — the pure-state core of
 * [AppForegroundRecycleHook]. Drives the lifecycle transitions
 * directly with a controllable clock so the threshold logic doesn't
 * need a real wall-clock wait.
 */
 class AppForegroundRecycleHookTest {
    @Test
    fun firstForegroundAfterProcessStartDoesNotPublish() {
        // The very first onActivityStarted has no prior background to
        // recycle from — recycling here would fire a redundant
        // re-handshake on every cold start, which is wasteful.
        var fakeNow = 0L
        var publishCount = 0
        val counter =
            AppForegroundCounter(
                publishEvent = { publishCount++ },
                nowMillis = { fakeNow },
            )
        fakeNow = 1_000L
        counter.onActivityStarted()
        assertEquals("first onActivityStarted must not publish — no prior background", 0, publishCount)
        assertEquals(0, counter.recyclesFired)
    }
    @Test
    fun shortBackgroundDoesNotTriggerRecycle() {
        // 1 s background (notification pull, biometric prompt) is
        // typical UI noise and the QUIC socket is still healthy.
        var fakeNow = 0L
        var publishCount = 0
        val counter =
            AppForegroundCounter(
                backgroundThresholdMs = 5_000L,
                publishEvent = { publishCount++ },
                nowMillis = { fakeNow },
            )
        fakeNow = 1_000L
        counter.onActivityStarted()
        fakeNow = 2_000L
        counter.onActivityStopped()
        fakeNow = 3_000L // backgrounded for 1 s only
        counter.onActivityStarted()
        assertEquals(
            "background < threshold must not publish — short transitions don't reclaim sockets",
            0,
            publishCount,
        )
    }
    @Test
    fun backgroundLongerThanThresholdTriggersRecycle() {
        // 6 s background crosses the 5 s default threshold — Android
        // may have reclaimed the socket FD by now, so recycle the
        // QUIC session on resume.
        var fakeNow = 0L
        var publishCount = 0
        val counter =
            AppForegroundCounter(
                backgroundThresholdMs = 5_000L,
                publishEvent = { publishCount++ },
                nowMillis = { fakeNow },
            )
        fakeNow = 1_000L
        counter.onActivityStarted()
        fakeNow = 2_000L
        counter.onActivityStopped()
        fakeNow = 8_000L // backgrounded for 6 s
        counter.onActivityStarted()
        assertEquals(
            "background ≥ threshold must publish exactly once on resume",
            1,
            publishCount,
        )
        assertEquals(1, counter.recyclesFired)
    }
    @Test
    fun multipleActivitiesTrackTransitionCorrectly() {
        // Picture-in-picture mode is implemented as a second activity
        // overlaid on the main activity. While both are started the
        // app is in foreground; only when both stop does the app
        // truly background.
        var fakeNow = 0L
        var publishCount = 0
        val counter =
            AppForegroundCounter(
                backgroundThresholdMs = 5_000L,
                publishEvent = { publishCount++ },
                nowMillis = { fakeNow },
            )
        // First activity starts (cold start), no publish.
        fakeNow = 1_000L
        counter.onActivityStarted()
        // Second activity (e.g. PIP / dialog) starts on top. Still
        // foreground; no publish — `wasBackgrounded` was false.
        fakeNow = 2_000L
        counter.onActivityStarted()
        // First activity stops (e.g. user backs out of main).
        // counter = 1, still foreground.
        fakeNow = 3_000L
        counter.onActivityStopped()
        assertEquals(
            "intermediate stop with another activity still started must not background",
            0,
            publishCount,
        )
        // Second stops → app truly backgrounds.
        fakeNow = 4_000L
        counter.onActivityStopped()
        // 6 s later, an activity restarts → recycle.
        fakeNow = 10_000L
        counter.onActivityStarted()
        assertEquals(1, publishCount)
    }
    @Test
    fun consecutiveLongBackgroundsEachPublishOnce() {
        // Two separate back-and-forth cycles must each fire exactly
        // one publish. A regression that misses to refresh the
        // last-backgrounded timestamp on second-stop would either
        // double-fire on the second resume or skip it.
        var fakeNow = 0L
        var publishCount = 0
        val counter =
            AppForegroundCounter(
                backgroundThresholdMs = 5_000L,
                publishEvent = { publishCount++ },
                nowMillis = { fakeNow },
            )
        // Cycle 1: cold start → 6 s background → resume (publish #1)
        fakeNow = 1_000L
        counter.onActivityStarted()
        fakeNow = 2_000L
        counter.onActivityStopped()
        fakeNow = 8_000L
        counter.onActivityStarted()
        assertEquals("first resume after long background must publish", 1, publishCount)
        // Cycle 2: 8 s background again → resume (publish #2)
        fakeNow = 10_000L
        counter.onActivityStopped()
        fakeNow = 18_000L
        counter.onActivityStarted()
        assertEquals("second resume after long background must also publish", 2, publishCount)
    }
 }
@@ -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 {
+    if (peek.keyPhase == conn.currentReceiveKeyPhase) {
-        peek.keyPhase == conn.currentReceiveKeyPhase -> {
+        parsed =
-            keysToUse = live
+            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
-        }
+        } else {
-
+            // Peer rotated (possibly for a 2nd time). Derive next-phase
-        // Reordered packet from before our last rotation — try the
+            // keys against the rolled-forward `appReceiveSecret` and
-        // retained previous keys. Reordering window is small but real
+            // try AEAD. On success we commit the rotation; on failure
-        // for paths with non-trivial RTT.
+            // it's a bona-fide corrupt / unauthenticated packet.
        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 -> {
            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",
@@ -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>()
            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<QuicConnection, InMemoryQuicPipe> =
        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 }
@@ -265,29 +265,45 @@ class KeyUpdatePeerInitiatedTest {
            assertEquals(QuicConnection.Status.CONNECTED, client.status)
        }
-    // KNOWN LIMITATION — consecutive rotations within the reorder window.
+    @Test
-    //
+    fun twoConsecutiveRotationsCommitCorrectly() =
-    // The parser currently routes inbound packets whose KEY_PHASE bit
+        runBlocking {
-    // does not match `currentReceiveKeyPhase` to
+            // Belt + braces: one rotation is the simple case; a second
-    // `previousReceiveProtection` whenever those prior keys exist. After
+            // rotation must derive off the FIRST-rotation secret, not
-    // a single rotation that's correct (matches RFC 9001 §6.1's reorder-
+            // the original handshake secret, and the parser must NOT
-    // tolerance contract). But after a SECOND rotation by the peer the
+            // misroute the second-rotation packet to
-    // KEY_PHASE bit wraps back to the original value — and our parser
+            // previousReceiveProtection (which would AEAD-fail and
-    // misroutes those packets to the (now-wrong) prior keys, AEAD-fails,
+            // silently drop, wedging the connection — the
-    // and silently drops them. The connection wedges.
+            // KNOWN-LIMITATION pre-fix this test pins now closes).
-    //
+            //
-    // The standard fix is to gate `previousReceiveProtection` on a
+            // The parser fix is "try previous keys; on AEAD failure
-    // packet-number threshold (track the PN of the rotation-triggering
+            // fall through to next-phase derivation" — neqo's
-    // packet; reroute to next-phase derivation once subsequent KEY_PHASE-
+            // approach. Two AEAD attempts on a mismatched-phase
-    // mismatched packets exceed it). neqo and picoquic implement this;
+            // packet are cheap; KEY_PHASE mismatch is rare to begin
-    // we don't yet. For audio-rooms' 3-hour session, ONE rotation is the
+            // with (at most once per a-billion-packets in normal
-    // realistic case (peers rotate sparingly), so the deeper fix is a
+            // usage).
-    // follow-up rather than a blocker.
+            val (client, pipe) = newConnectedClient()
-    //
+
-    // No test asserts the broken behaviour — adding one would just pin
+            pipe.rotateServerApplicationKeys()
-    // the regression. When the protocol-level fix lands, a positive
+            feedDatagram(client, pipe.buildServerApplicationDatagram(listOf(PingFrame))!!, nowMillis = 0L)
-    // `consecutiveRotationsCommitCorrectly` test is the right addition
+            assertEquals(true, client.currentReceiveKeyPhase, "first rotation must flip the bit")
-    // here.
+
            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
@@ -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<Long>()
            for (id in streamIds) {
                if (rng.nextDouble() < lossRate) {
                    droppedIds += id
                    continue
                }
                val packet =
                    pipe.buildServerApplicationDatagram(
                        listOf(StreamFrame(streamId = id, offset = 0L, data = payloads[id]!!, fin = true)),
                    )!!
                feedDatagram(client, packet, nowMillis = 0L)
            }
            val received = mutableListOf<Long>()
            for (id in streamIds) {
                if (id in droppedIds) continue
                val s = client.streamById(id) ?: continue
                if (withTimeoutOrNull(2_000L) { s.incoming.toList() } != null) {
                    received += id
                }
            }
            // Expected delivery is approximately (1 - lossRate) of the
            // total. RNG variance gives ~5% tolerance band; we assert
            // ≥ 60% to leave plenty of margin while still catching a
            // catastrophic collapse (e.g. parser silently drops every
            // packet after the first loss).
            val floor = (groupCount * 0.6).toInt()
            assertTrue(
                received.size >= floor,
                "listener received ${received.size} / $groupCount under 20% loss — floor $floor. " +
                    "dropped=${droppedIds.size}",
            )
            assertEquals(
                QuicConnection.Status.CONNECTED,
                client.status,
                "connection must stay CONNECTED through extreme loss",
            )
        }
    @Test
    fun reliableBidiStreamRecoversFromMidStreamPacketLoss() =
        runBlocking {
            // Reliable-stream loss harness — distinct from the moq-
            // lite group-stream best-effort path above. RFC 9000
            // STREAM frames carry full reliability semantics: a
            // dropped packet must trigger retransmit on PTO, and
            // the listener MUST eventually see the bytes contiguous
            // and in order. This test pins that contract end-to-end
            // by:
            //   1. Server sends 4 STREAM frames on the same bidi
            //      stream, each in its own datagram, covering a
            //      monotonic offset range.
            //   2. The middle two datagrams are dropped on first
            //      delivery — simulating a transient mid-stream
            //      loss.
            //   3. Server retransmits the dropped ranges on the
            //      same offsets (mimicking what RFC 9002
            //      retransmit logic does on PTO).
            //   4. Client must surface all 4 chunks contiguous,
            //      in order, no gaps.
            val (client, pipe) = newConnectedClient()
            // Open a client-bidi stream so the server can write back.
            val stream = client.openBidiStream()
            val streamId = stream.streamId
            val chunks =
                listOf(
                    "AAAA".encodeToByteArray(),
                    "BBBB".encodeToByteArray(),
                    "CCCC".encodeToByteArray(),
                    "DDDD".encodeToByteArray(),
                )
            val offsets = LongArray(chunks.size)
            var off = 0L
            for ((i, c) in chunks.withIndex()) {
                offsets[i] = off
                off += c.size
            }
            // First-pass delivery: drop chunks[1] and chunks[2].
            for (i in chunks.indices) {
                if (i == 1 || i == 2) continue
                val frame =
                    StreamFrame(
                        streamId = streamId,
                        offset = offsets[i],
                        data = chunks[i],
                        fin = i == chunks.size - 1,
                    )
                feedDatagram(client, pipe.buildServerApplicationDatagram(listOf(frame))!!, nowMillis = 0L)
            }
            // Retransmit the dropped chunks on their original offsets.
            for (i in listOf(1, 2)) {
                val frame =
                    StreamFrame(
                        streamId = streamId,
                        offset = offsets[i],
                        data = chunks[i],
                        fin = false,
                    )
                feedDatagram(client, pipe.buildServerApplicationDatagram(listOf(frame))!!, nowMillis = 0L)
            }
            val collected = withTimeoutOrNull(2_000L) { stream.incoming.toList() }
            assertNotNull(collected, "stream must complete after retransmits fill the gaps")
            val joined = ByteArray(collected.sumOf { it.size })
            var p = 0
            for (c in collected) {
                c.copyInto(joined, p)
                p += c.size
            }
            assertEquals(
                "AAAABBBBCCCCDDDD",
                joined.decodeToString(),
                "reliable bidi stream must surface every byte in offset order even after " +
                    "mid-stream packet loss + retransmit",
            )
            assertEquals(QuicConnection.Status.CONNECTED, client.status)
        }
    private fun newConnectedClient(): Pair<QuicConnection, InMemoryQuicPipe> =
        runBlocking {
            val client =