feat(quic): retire fully-settled streams to keep tracker bounded under audio-room churn

Soak target #1 from the audio-rooms hardening pass: moq-lite over QUIC
mints one peer-uni stream per Opus frame, so a 3-hour broadcast at
~50 frames/sec accumulated ~540 000 stream entries in
`QuicConnection.streamsList` / `streams` for the lifetime of the
session. The two structures were append-only — closed streams were
filtered out of the writer's iteration but never removed — and the
heap grew monotonically.

Adds `QuicStream.isFullyRetired` plus `retireFullyDoneStreamsLocked`
on the connection. The writer drains the retire pass at the top of
`buildApplicationPacket`, dropping streams whose send side has
peer-acked FIN/RESET and whose receive side has both FIN'd and
fully drained into the application's incoming Channel. The
cumulative receive high-water folds into `retiredStreamsRecvBytes`
so the connection-level MAX_DATA accounting in
`appendFlowControlUpdates` keeps advertising the lifetime total —
without that seed, retiring K bytes would silently regress the
peer's send credit.

Also adds soak target #6 coverage: `QuicConnectionDriver` now
exposes `driverJob` / `closeTeardownJob` for test assertion, and
the new `QuicConnectionDriverLifecycleTest` cycles 100 sessions
against a localhost UDP blackhole to pin idempotent close +
bounded thread growth.

Tests:
 - `StreamRetirementSoakTest` (4 cases): local-uni FIN+ACK
   retirement, peer-uni listener-path retirement, MAX_DATA accounting
   preservation across retire, and a 10 000-stream churn harness
   that asserts the working set stays bounded.
 - `QuicConnectionDriverLifecycleTest` (2 cases): close idempotency
   and 100-session thread-leak canary.

https://claude.ai/code/session_018KPKWRg5baX5Anf7zfEyec

quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt

@@ -301,15 +301,55 @@ class QuicConnection(
     /**
      * Round-4 perf #10: parallel insertion-ordered list of streams so the
      * writer's round-robin scan can index by position without
-     * `streams.entries.toList()` allocating per drain. Streams are only ever
+     * `streams.entries.toList()` allocating per drain. Mutated in lockstep
-     * added (no removal in the current model), so the two stay in sync as
+     * with [streams] under [streamsLock]:
-     * long as `getOrCreatePeerStreamLocked` and `openBidi/UniStream` append
+     *  - [openBidiStreamLocked] / [openUniStreamLocked] /
-     * to both.
+     *    [getOrCreatePeerStreamLocked] append.
+     *  - [retireFullyDoneStreamsLocked] removes entries whose stream has
+     *    flipped [QuicStream.isFullyRetired] = true. The retire pass
+     *    folds the receive-side high-water mark into
+     *    [retiredStreamsRecvBytes] so the writer's MAX_DATA accounting
+     *    in `appendFlowControlUpdates` doesn't regress when a retired
+     *    stream's `receive.contiguousEnd()` drops out of the iteration.
+     *
+     * Without retirement, the moq-lite audio-rooms path leaks one
+     * QuicStream per Opus frame for the lifetime of the session — the
+     * stream-cliff investigation in
+     * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`
+     * pegs steady-state churn at ~50 streams/sec, so a 3-hour room
+     * accumulates ~540 000 entries before retirement was wired.
      */
     private val streamsList = mutableListOf<QuicStream>()
     private var nextLocalBidiIndex: Long = 0L
     private var nextLocalUniIndex: Long = 0L
 
+    /**
+     * Cumulative `receive.contiguousEnd()` from streams that have been
+     * removed by [retireFullyDoneStreamsLocked]. Folded into the writer's
+     * `totalRecvAdvanced` accumulator so MAX_DATA continues to advertise
+     * the *connection-lifetime* receive high-water mark even after the
+     * contributing streams have been dropped from [streamsList]. Without
+     * this, retiring N streams that each delivered K bytes would silently
+     * regress the advertised connection-level credit by N*K, eventually
+     * starving the peer once `advertisedMaxData` was tripped past.
+     *
+     * Caller of any read/write must hold [streamsLock].
+     */
+    internal var retiredStreamsRecvBytes: Long = 0L
+        private set
+
+    /**
+     * Cumulative count of streams ever removed by
+     * [retireFullyDoneStreamsLocked]. Diagnostic-only; tests that drive
+     * stream churn use this to assert retirement actually fired (rather
+     * than relying on `streamsList.size` alone, which can shrink because
+     * a soak loop happened to drain the same workload it just enqueued).
+     *
+     * Caller of any read/write must hold [streamsLock].
+     */
+    internal var retiredStreamsCount: Long = 0L
+        private set
+
     /**
      * Peer-advertised concurrent bidirectional stream cap. Initialised from
      * [TransportParameters.initialMaxStreamsBidi] when peer params arrive,
@@ -1685,11 +1725,75 @@ class QuicConnection(
     /**
      * Insertion-ordered list view used by the writer's round-robin scan.
      * Stays in sync with [streams] because the only mutation paths
-     * (openBidi/UniStream, getOrCreatePeerStreamLocked) append to both. No
+     * (openBidi/UniStream, getOrCreatePeerStreamLocked,
-     * remove path exists today; if/when one is added it MUST update both.
+     * retireFullyDoneStreamsLocked) update both.
      */
     internal fun streamsListLocked(): List<QuicStream> = streamsList
 
+    /**
+     * Walk [streamsList] once and remove every stream whose
+     * [QuicStream.isFullyRetired] flag is true. Drops the entry from both
+     * [streamsList] and [streams]. Receive-side high-water marks fold
+     * into [retiredStreamsRecvBytes] so the writer's connection-level
+     * MAX_DATA accounting continues to see the lifetime total.
+     *
+     * Returns the number of streams removed in this pass.
+     *
+     * Caller MUST hold [streamsLock]. The writer drains under
+     * `streamsLock`, so calling this from the top of `buildApplicationPacket`
+     * is the natural place — it runs once per send pass, sees the latest
+     * FIN/RESET ACK state from the parser's just-finished processing, and
+     * the iteration order matches the writer's per-pass round-robin
+     * (so the rotation cursor doesn't accidentally point at a hole).
+     *
+     * Why retirement is safe even though the QUIC spec requires the peer
+     * to deliver retransmits if its ACK never reached us:
+     *  - SEND side waits for `finAcked` / `resetAcked`, i.e. the peer has
+     *    already confirmed receipt of our FIN. After that point the peer
+     *    will not re-send anything on the stream.
+     *  - RECEIVE side waits for the parser to have pushed every byte to
+     *    the application's incoming Channel ([ReceiveBuffer.isFullyRead]).
+     *    Subsequent retransmits from the peer (rare — peer only retransmits
+     *    if its own loss-detection fires) would land on a fresh stream
+     *    object created by [getOrCreatePeerStreamLocked], deliver
+     *    duplicate bytes, and re-fire FIN to a closed channel; the moq-lite
+     *    listener treats duplicates as no-ops, so the failure mode is "a
+     *    bit of wasted CPU" rather than "wrong audio". The alternative
+     *    (retain every stream forever) is a hard memory leak in soak.
+     */
+    internal fun retireFullyDoneStreamsLocked(): Int {
+        if (streamsList.isEmpty()) return 0
+        var removed = 0
+        val it = streamsList.iterator()
+        while (it.hasNext()) {
+            val stream = it.next()
+            if (!stream.isFullyRetired) continue
+            // Fold the per-stream receive high-water into the cumulative
+            // counter BEFORE we drop it — once we lose the reference the
+            // writer can no longer reconstruct the contribution.
+            retiredStreamsRecvBytes += stream.receive.contiguousEnd()
+            // Defence-in-depth: ensure the application-side incoming
+            // channel is closed even if the parser somehow missed the
+            // closeIncoming call (e.g. a stream that finished via
+            // resetAcked never having received any STREAM frame at all).
+            // closeIncoming is idempotent.
+            stream.closeIncoming()
+            streams.remove(stream.streamId)
+            it.remove()
+            removed++
+        }
+        if (removed > 0) {
+            retiredStreamsCount += removed.toLong()
+            // The writer's round-robin cursor is a position in
+            // [streamsList], so a removal that crossed the cursor would
+            // make the next drain pass skip a tier. Reset to 0 — the
+            // round-robin is just a fairness hint, not a semantic
+            // requirement.
+            streamRoundRobinStart = 0
+        }
+        return removed
+    }
+
     /** Caller must hold [lock]. Pending datagram queue for the driver's send loop. */
     internal fun pendingDatagramsLocked(): ArrayDeque<ByteArray> = pendingDatagrams
 

quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt

@@ -64,6 +64,30 @@ class QuicConnectionDriver(
     private var readJob: Job? = null
     private var sendJob: Job? = null
 
+    /**
+     * Test-only handle on the driver's [SupervisorJob]. Used by the
+     * session-lifecycle leak test in
+     * [com.vitorpamplona.quic.connection.QuicConnectionDriverLifecycleTest]
+     * to assert that a closed driver reports `isCompleted = true` once
+     * its read/send loops have unwound — the inverse of "the driver
+     * leaked a coroutine past close()".
+     *
+     * Production code MUST NOT touch this — the driver lifecycle is
+     * managed end-to-end by [start] / [close].
+     */
+    internal val driverJob: Job get() = job
+
+    /**
+     * Test-only handle on the in-flight teardown coroutine. Returns null
+     * before [close] has been called; once close runs, the returned Job
+     * lets the test `join()` until teardown is complete (cancel + socket
+     * close + read/send join). Pre-existing close() returned immediately
+     * and provided no synchronous "teardown is done" signal — tests had
+     * to poll `connection.status == CLOSED` and trust that the rest of
+     * the cleanup eventually settled.
+     */
+    internal val closeTeardownJob: Job? get() = closeJob
+
     /**
      * Round-5 concurrency #5: close() guard. A second concurrent invocation
      * (e.g. session close + read-loop death close racing) used to launch a

quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt

@@ -575,6 +575,18 @@ private fun buildApplicationPacket(
     // ServerHello has been processed.
     val use1Rtt = state.sendProtection != null
     val proto = state.sendProtection ?: conn.zeroRttSendProtection ?: return null
+    // Drop fully-settled streams BEFORE iterating so we don't waste a
+    // round-robin slot on a stream with nothing to send and no receive
+    // bookkeeping pending. Run this once per drain (at the top of the
+    // application-level path, the only level where streams exist) under
+    // the same `streamsLock` the rest of the drain holds — see
+    // [QuicConnection.retireFullyDoneStreamsLocked] for the safety
+    // argument. The cumulative receive bytes from removed streams fold
+    // into `appendFlowControlUpdates` below so MAX_DATA stays correct.
+    // Safe to call on the 0-RTT path too: no stream can be fully
+    // settled mid-handshake (peer hasn't ACK'd anything yet), so the
+    // walk is a no-op.
+    conn.retireFullyDoneStreamsLocked()
     val frames = mutableListOf<Frame>()
     // Tokens collected in lock-step with [frames]: each retransmittable
     // frame contributes a [RecoveryToken] so the [SentPacket] recorded
@@ -957,7 +969,15 @@ private fun appendFlowControlUpdates(
     }
 
     val cfg = conn.config
-    var totalRecvAdvanced = 0L
+    // Seed with the cumulative receive high-water mark from streams that
+    // [QuicConnection.retireFullyDoneStreamsLocked] has already dropped
+    // — the writer's connection-level MAX_DATA threshold uses the
+    // *lifetime* total, so retired streams must keep contributing even
+    // after their per-object `receive.contiguousEnd()` is no longer
+    // reachable. Without this seed, retiring K bytes of streams would
+    // silently regress the advertised credit by K and eventually starve
+    // the peer once the running total fell behind `advertisedMaxData`.
+    var totalRecvAdvanced = conn.retiredStreamsRecvBytes
     // Round-4 perf #9 + round-5 #9: walk the streams via the index-friendly
     // list view (no `entries.toList()` allocation), and only do per-stream
     // window/threshold work for streams flagged by the parser since the last

quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt

@@ -113,6 +113,64 @@ class QuicStream(
     val isClosed: Boolean
         get() = send.finSent && receive.finReceived
 
+    /**
+     * True once both directions are *fully* settled and the stream may be
+     * removed from the connection's tracking lists (see
+     * `QuicConnection.retireFullyDoneStreamsLocked`). This is strictly
+     * stronger than [isClosed]: the latter only requires the FIN bits to
+     * have been observed, not that the peer has acknowledged our FIN /
+     * RESET (send side) nor that the application has drained the buffered
+     * receive bytes (receive side).
+     *
+     * Lifetime contract:
+     *  - SEND side done: peer has ACK'd our FIN ([SendBuffer.finAcked]) OR
+     *    we ABORTed the stream and the peer ACK'd our RESET_STREAM
+     *    ([resetAcked]). For [Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL]
+     *    there is no send side, so this leg is trivially done.
+     *  - RECEIVE side done: peer FIN'd ([ReceiveBuffer.finReceived]) AND
+     *    every byte has been delivered from the receive buffer to the
+     *    application's incoming Channel ([ReceiveBuffer.isFullyRead]).
+     *    Once both hold, the parser has already invoked [closeIncoming]
+     *    so any application coroutine still draining the buffered
+     *    [incoming] flow will terminate naturally — the QuicStream object
+     *    can outlive its membership in the connection's `streamsList`.
+     *    For [Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE] there is no
+     *    receive side, so this leg is trivially done.
+     *
+     * The motivation is the moq-lite audio-rooms path: each Opus frame is
+     * forwarded as a fresh peer-uni stream by the relay. A 3-hour session
+     * at ~50 frames/sec churns ~540 000 streams. Without retirement,
+     * `streamsList` and the `streams` map grow monotonically — the
+     * `nestsClient/plans/2026-04-26-moq-lite-gap.md` soak target wants
+     * memory flat past handshake-stable. Stream retirement is the only
+     * QUIC-level fix that keeps the tracker bounded for that workload.
+     *
+     * Read this from any thread under the connection's `streamsLock` — the
+     * underlying flags are `@Volatile` and the buffers' synchronized
+     * blocks publish their state atomically.
+     */
+    val isFullyRetired: Boolean
+        get() {
+            val sendSettled =
+                when (direction) {
+                    Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL -> true
+
+                    Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE,
+                    Direction.BIDIRECTIONAL,
+                    -> send.finAcked || resetAcked
+                }
+            if (!sendSettled) return false
+            val recvSettled =
+                when (direction) {
+                    Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE -> true
+
+                    Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL,
+                    Direction.BIDIRECTIONAL,
+                    -> receive.finReceived && receive.isFullyRead()
+                }
+            return recvSettled
+        }
+
     /**
      * Pushes [data] toward the consumer. Returns false if the bounded channel
      * was full; the caller (parser) is expected to escalate to a connection-

quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetirementSoakTest.kt

@@ -0,0 +1,410 @@
+/*
+ * 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.flow.toList
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * Pins the stream-retirement contract that keeps `streamsList` /
+ * `streams` bounded across long-lived audio-room sessions.
+ *
+ * Production motivation: the moq-lite listener path in
+ * `nestsClient/src/commonMain/kotlin/.../moq/lite/` receives one
+ * peer-uni stream per Opus frame. At ~50 frames/sec a 3-hour broadcast
+ * mints ~540 000 streams. Pre-retirement, every stream stayed pinned in
+ * `streamsList` (insertion-only) and `streams` (no remove) for the
+ * lifetime of the connection — which made `QuicConnection`'s heap grow
+ * monotonically until the audio room session was torn down. The
+ * acceptance criterion in the soak prompt is "no monotonic growth past
+ * handshake-stable"; this file is the unit-test surface that pins the
+ * mechanics behind that property.
+ *
+ * Three angles + a soak-shape:
+ *  1. [retiredLocalUniStreamsAreRemovedAfterFinAndAck] — local-uni send
+ *     path: client opens uni streams, writes + FIN, peer ACKs. Once FIN
+ *     is `markAcked`'d, the next writer drain retires them.
+ *  2. [retiredPeerInitiatedStreamsDoNotPinReceiveBuffers] — moq-lite
+ *     listener path: peer opens server-uni streams, sends payload + FIN,
+ *     parser drains contiguous bytes and closes incoming. Next writer
+ *     drain retires them.
+ *  3. [retirementPreservesConnectionLevelMaxDataAccounting] — the subtle
+ *     correctness check. Removing N retired streams must NOT regress the
+ *     writer's connection-level MAX_DATA accounting; the
+ *     `retiredStreamsRecvBytes` accumulator on the connection folds
+ *     per-stream `receive.contiguousEnd()` sums forward. After retire,
+ *     `advertisedMaxData` continues to grow on subsequent receive bytes.
+ *  4. [streamChurnSoakKeepsTrackerBounded] — soak-shape: cycle through
+ *     200 generations of 50 streams (10 000 stream lifecycles, well past
+ *     steady-state) and assert the working set stays bounded. Without
+ *     retirement the working set would equal totalStreams.
+ */
+class StreamRetirementSoakTest {
+    @Test
+    fun retiredLocalUniStreamsAreRemovedAfterFinAndAck() =
+        runBlocking {
+            val (client, pipe) = newConnectedClient()
+            val n = 50
+
+            val streams =
+                client.openUniStreamsBatch(items = (0 until n).toList()) { stream, i ->
+                    stream.send.enqueue(
+                        "frame-${i.toString().padStart(2, '0')}".encodeToByteArray(),
+                    )
+                    stream.send.finish()
+                    stream
+                }
+            assertEquals(n, client.streamsListLocked().size)
+
+            // Drain every outbound packet the writer has to send.
+            val anyOutput = drainAll(client, pipe)
+            assertTrue(anyOutput, "writer must have emitted at least one application packet")
+
+            // Peer ACKs every PN we ever sent on the application space.
+            // largestAcked = nextPacketNumber - 1; firstAckRange covers
+            // PN 0 through largestAcked inclusive.
+            val largestSent = client.application.pnSpace.nextPacketNumber - 1L
+            assertTrue(
+                largestSent >= 0L,
+                "drainAll must have allocated at least one application PN (got=$largestSent)",
+            )
+            val ackPacket =
+                pipe.buildServerApplicationDatagram(
+                    listOf(
+                        AckFrame(
+                            largestAcknowledged = largestSent,
+                            ackDelay = 0L,
+                            firstAckRange = largestSent,
+                        ),
+                    ),
+                )!!
+            feedDatagram(client, ackPacket, nowMillis = 0L)
+
+            for ((idx, s) in streams.withIndex()) {
+                assertTrue(
+                    s.send.finAcked,
+                    "stream[$idx] (id=${s.streamId}) finAcked must latch after peer ACK",
+                )
+                assertTrue(
+                    s.isFullyRetired,
+                    "stream[$idx] uni-out direction is fully retired after finAcked",
+                )
+            }
+
+            // One more drain triggers retireFullyDoneStreamsLocked at the
+            // top of buildApplicationPacket — the production seam.
+            drainAll(client, pipe)
+
+            assertEquals(
+                0,
+                client.streamsListLocked().size,
+                "streamsList must be empty after retirement of all FIN-acked uni streams",
+            )
+            assertEquals(
+                n.toLong(),
+                client.retiredStreamsCount,
+                "retiredStreamsCount must equal the number of retired streams",
+            )
+            assertEquals(
+                0L,
+                client.retiredStreamsRecvBytes,
+                "uni-out streams contribute no receive-side bytes to the accumulator",
+            )
+        }
+
+    @Test
+    fun retiredPeerInitiatedStreamsDoNotPinReceiveBuffers() =
+        runBlocking {
+            val (client, pipe) = newConnectedClient()
+            val n = 30
+
+            val firstServerUniId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
+            val serverUniIds = (0 until n).map { firstServerUniId + 4L * it }
+            val payloads = serverUniIds.associateWith { id -> "g-$id".encodeToByteArray() }
+
+            for (id in serverUniIds) {
+                val frame =
+                    StreamFrame(
+                        streamId = id,
+                        offset = 0L,
+                        data = payloads[id]!!,
+                        fin = true,
+                    )
+                val packet = pipe.buildServerApplicationDatagram(listOf(frame))!!
+                feedDatagram(client, packet, nowMillis = 0L)
+            }
+
+            // Every per-stream incoming Flow must complete (parser closed
+            // it via the isFullyRead branch).
+            for (id in serverUniIds) {
+                val stream = client.streamById(id)!!
+                val chunks = withTimeoutOrNull(2_000L) { stream.incoming.toList() }
+                assertTrue(chunks != null, "stream $id incoming Flow must terminate after FIN")
+                val joined = ByteArray(chunks.sumOf { it.size })
+                var p = 0
+                for (c in chunks) {
+                    c.copyInto(joined, p)
+                    p += c.size
+                }
+                assertEquals(
+                    payloads[id]!!.decodeToString(),
+                    joined.decodeToString(),
+                    "stream $id payload must surface intact",
+                )
+            }
+
+            // Drain → retireFullyDoneStreamsLocked at the top of
+            // buildApplicationPacket drops every server-uni stream:
+            //   - direction = UNIDIRECTIONAL_REMOTE_TO_LOCAL → send
+            //     side is trivially settled
+            //   - receive.finReceived AND receive.isFullyRead() because
+            //     the parser drained every chunk into the incoming
+            //     channel before closing it
+            drainAll(client, pipe)
+
+            assertEquals(
+                0,
+                client.streamsListLocked().size,
+                "every server-uni stream must be retired after parsing FIN + writer drain",
+            )
+            assertEquals(
+                n.toLong(),
+                client.retiredStreamsCount,
+                "retiredStreamsCount must equal the number of peer-uni streams retired",
+            )
+            val expectedRecvBytes = payloads.values.sumOf { it.size }.toLong()
+            assertEquals(
+                expectedRecvBytes,
+                client.retiredStreamsRecvBytes,
+                "retiredStreamsRecvBytes must aggregate every retired stream's receive high-water",
+            )
+        }
+
+    @Test
+    fun retirementPreservesConnectionLevelMaxDataAccounting() =
+        runBlocking {
+            // Without the retiredStreamsRecvBytes seed, the writer's
+            // `totalRecvAdvanced` would reset to 0 after retirement and
+            // its half-window MAX_DATA threshold check would never fire
+            // again — the peer's send credit would silently freeze at
+            // initialMaxData. Pin the seed by:
+            //  1. Pushing enough peer bytes to exceed initialMaxData / 2,
+            //     forcing a fresh MAX_DATA frame on the wire.
+            //  2. Triggering retirement.
+            //  3. Pushing a second wave that ALSO crosses the half-window
+            //     threshold from the *post-retire* baseline.
+            // If the writer's seed is wrong, step 3 fails to advance
+            // advertisedMaxData and MaxDataFrame is not emitted.
+            val (client, pipe) = newConnectedClient()
+            val cfg = client.config
+            val perStream = (cfg.initialMaxStreamDataUni / 4).coerceAtMost(8L * 1024)
+            val streamCountForFirstWave =
+                (cfg.initialMaxData / 2 / perStream + 1).toInt().coerceAtLeast(2)
+
+            val firstStart = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
+            for (i in 0 until streamCountForFirstWave) {
+                val id = firstStart + 4L * i
+                val payload = ByteArray(perStream.toInt()) { (it and 0xFF).toByte() }
+                val packet =
+                    pipe.buildServerApplicationDatagram(
+                        listOf(StreamFrame(streamId = id, offset = 0L, data = payload, fin = true)),
+                    )!!
+                feedDatagram(client, packet, nowMillis = 0L)
+                client.streamById(id)?.incoming?.toList()
+            }
+
+            // Drain so MAX_DATA goes out + retirement runs.
+            drainAll(client, pipe)
+            val advertisedAfterFirstWave = client.advertisedMaxData
+            assertTrue(
+                advertisedAfterFirstWave > cfg.initialMaxData,
+                "first wave must have bumped advertisedMaxData (was=$advertisedAfterFirstWave " +
+                    "initialMaxData=${cfg.initialMaxData}) — fixture pre-condition for the " +
+                    "retirement-preserves-credit test",
+            )
+            assertEquals(
+                0,
+                client.streamsListLocked().size,
+                "first wave must be fully retired before the second wave begins",
+            )
+
+            // Second wave — push enough additional bytes that the running
+            // total (retired + open) crosses the next half-window
+            // threshold. Without the seed in totalRecvAdvanced, the
+            // second wave would not raise advertisedMaxData.
+            val secondStart = firstStart + 4L * streamCountForFirstWave
+            for (i in 0 until streamCountForFirstWave) {
+                val id = secondStart + 4L * i
+                val payload = ByteArray(perStream.toInt()) { ((it + 1) and 0xFF).toByte() }
+                val packet =
+                    pipe.buildServerApplicationDatagram(
+                        listOf(StreamFrame(streamId = id, offset = 0L, data = payload, fin = true)),
+                    )!!
+                feedDatagram(client, packet, nowMillis = 0L)
+                client.streamById(id)?.incoming?.toList()
+            }
+            drainAll(client, pipe)
+
+            val advertisedAfterSecondWave = client.advertisedMaxData
+            assertTrue(
+                advertisedAfterSecondWave > advertisedAfterFirstWave,
+                "second wave must continue to advance advertisedMaxData past " +
+                    "$advertisedAfterFirstWave (was $advertisedAfterSecondWave) — if it didn't, " +
+                    "retirement regressed the writer's connection-level MAX_DATA accounting",
+            )
+            // The cumulative receive total — folded into the writer's
+            // totalRecvAdvanced via `retiredStreamsRecvBytes` — must
+            // include every stream from both waves.
+            val expectedCumulativeRecv = 2L * streamCountForFirstWave * perStream
+            assertTrue(
+                client.retiredStreamsRecvBytes >= expectedCumulativeRecv,
+                "retiredStreamsRecvBytes must include both waves; got=${client.retiredStreamsRecvBytes} " +
+                    "expected≥$expectedCumulativeRecv",
+            )
+        }
+
+    @Test
+    fun streamChurnSoakKeepsTrackerBounded() =
+        runBlocking {
+            // Soak-shape: simulate the moq-lite listener over many group
+            // generations. 200 generations × 50 streams = 10 000 stream
+            // lifecycles. Without retirement, streamsList would hold all
+            // 10 000 entries. With retirement, the working set stays
+            // bounded at the per-generation batch.
+            val (client, pipe) = newConnectedClient()
+            val perGen = 50
+            val gens = 200
+            val totalStreams = perGen * gens
+            var maxLiveStreams = 0
+            var nextStreamId = StreamId.build(StreamId.Kind.SERVER_UNI, 0L)
+
+            for (g in 0 until gens) {
+                for (i in 0 until perGen) {
+                    val payload = ByteArray(8) { ((g * perGen + i) and 0xFF).toByte() }
+                    val packet =
+                        pipe.buildServerApplicationDatagram(
+                            listOf(StreamFrame(streamId = nextStreamId, offset = 0L, data = payload, fin = true)),
+                        )!!
+                    feedDatagram(client, packet, nowMillis = 0L)
+                    client.streamById(nextStreamId)?.incoming?.toList()
+                    nextStreamId += 4L
+                }
+                val live = client.streamsListLocked().size
+                if (live > maxLiveStreams) maxLiveStreams = live
+                drainAll(client, pipe)
+            }
+
+            assertEquals(
+                0,
+                client.streamsListLocked().size,
+                "after the final retirement pass streamsList must drain fully",
+            )
+            assertEquals(
+                totalStreams.toLong(),
+                client.retiredStreamsCount,
+                "retiredStreamsCount must equal every stream the soak ever opened",
+            )
+            // Bound: working set should stay near per-generation batch.
+            // If retirement regressed (e.g. only fired at the end), we
+            // would observe the full totalStreams here.
+            assertTrue(
+                maxLiveStreams <= 2 * perGen,
+                "tracker working set must stay bounded at ~$perGen but observed $maxLiveStreams " +
+                    "across $gens generations — retirement is leaking",
+            )
+        }
+
+    /**
+     * Drain every outbound application packet the writer has to send.
+     * Returns true if at least one was emitted. The pipe consumes them
+     * (decrypting each one updates its inbound PN tracking) but does
+     * not itself reply — these tests inject ACKs explicitly.
+     */
+    private fun drainAll(
+        client: QuicConnection,
+        pipe: InMemoryQuicPipe,
+    ): Boolean {
+        var any = false
+        while (true) {
+            val out = drainOutbound(client, nowMillis = 0L) ?: break
+            any = true
+            // Decrypting bumps the pipe's applicationPnSpace so future
+            // server-built packets carry an up-to-date `largestAckedInSpace`
+            // value (avoids the truncated-PN encoder underflow that
+            // otherwise breaks long sequences).
+            pipe.decryptClientApplicationFrames(out)
+        }
+        return any
+    }
+
+    private fun newConnectedClient(): Pair<QuicConnection, InMemoryQuicPipe> =
+        runBlocking {
+            val client =
+                QuicConnection(
+                    serverName = "example.test",
+                    config =
+                        QuicConnectionConfig(
+                            initialMaxStreamsBidi = 4096,
+                            initialMaxStreamsUni = 65_536,
+                            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 = 4096,
+                            initialMaxStreamsUni = 65_536,
+                            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
+        }
+}

quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriverLifecycleTest.kt

@@ -0,0 +1,220 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quic.connection
+
+import com.vitorpamplona.quic.tls.PermissiveCertificateValidator
+import com.vitorpamplona.quic.transport.UdpSocket
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
+import java.net.DatagramSocket
+import java.net.InetSocketAddress
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * Pins the driver close-path cleanup contract for soak target #6:
+ * "Resource cleanup at session close. Run a session, close it, run
+ * another, repeat 100×. Verify no thread leaks ([Thread.getAllStackTraces]
+ * count stable), no socket leaks (netstat count stable), no
+ * Dispatchers.IO worker buildup."
+ *
+ * The driver spawns three things that need clean teardown:
+ *  - A [kotlinx.coroutines.SupervisorJob] parented to the test's scope
+ *    (driver.driverJob).
+ *  - Two child coroutines (read loop + send loop) launched on
+ *    `parentScope.coroutineContext + job + Dispatchers.IO`.
+ *  - A [com.vitorpamplona.quic.transport.UdpSocket] bound to an
+ *    ephemeral OS port.
+ *
+ * If any of these leaks across [QuicConnectionDriver.close], a 3-hour
+ * audio-room session that the user joins+leaves repeatedly (typical
+ * UX: switch rooms a few times, the OS swaps networks, etc.) eventually
+ * exhausts file descriptors or blooms the JVM thread pool.
+ *
+ * The fixture uses a localhost UDP "blackhole" — a DatagramSocket bound
+ * to an ephemeral port that does NOT speak QUIC. The client driver
+ * tries to handshake but never succeeds; the test isn't about
+ * handshake correctness, it's about what happens to the driver's
+ * resources when the application aborts the session before connect.
+ *
+ * Acceptance bands (intentionally generous to avoid CI flakiness):
+ *  - Per-session: connection.status latches CLOSED inside close()'s
+ *    bounded wait, driver.driverJob.isCompleted == true after we
+ *    cancel the parent scope, and a second close() call is a no-op
+ *    (idempotency contract added in round-5 #5).
+ *  - Across 100 sessions: net thread growth < 16, where the typical
+ *    Dispatchers.IO worker pool fluctuation is 4–8 on this JVM. A
+ *    leak that creates one persistent thread per session would
+ *    produce ~100 net growth; 16 is two orders of magnitude below
+ *    that and well above ambient JVM noise.
+ */
+class QuicConnectionDriverLifecycleTest {
+    private val blackhole: DatagramSocket = DatagramSocket(InetSocketAddress("127.0.0.1", 0))
+
+    @AfterTest
+    fun tearDown() {
+        runCatching { blackhole.close() }
+    }
+
+    @Test
+    fun closeIsIdempotentAndDriverJobCompletes() =
+        runBlocking {
+            val parent = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+            val (driver, connection, socket) = startDriver(parent)
+
+            // Let the driver kick off a few PTO cycles so there's actual
+            // in-flight crypto when close() runs — exercise the
+            // close-flush path that actually has work to flush.
+            delay(50)
+
+            driver.close()
+            // close() launches a teardown coroutine on parentScope and
+            // returns immediately. Wait for that coroutine to finish.
+            withTimeoutOrNull(5_000L) { driver.closeTeardownJob?.join() }
+                ?: error("close teardown coroutine never completed within 5 s — driver leaked")
+
+            assertEquals(
+                QuicConnection.Status.CLOSED,
+                connection.status,
+                "connection.status must be CLOSED after the close-teardown coroutine joins",
+            )
+            assertTrue(
+                socketIsClosed(socket),
+                "socket.close() must have run by the time the teardown coroutine completes",
+            )
+
+            // Idempotency: a second close() must be a no-op (round-5 #5
+            // memoizes the teardown launch). It must not throw, must
+            // not relaunch teardown, must not produce a new
+            // closeTeardownJob — the original Job stays.
+            val originalTeardown = driver.closeTeardownJob
+            driver.close()
+            assertTrue(
+                driver.closeTeardownJob === originalTeardown,
+                "second close() must reuse the memoized teardown Job (round-5 #5 idempotency)",
+            )
+
+            // Cancel the parent so SupervisorJob children unwind. Then
+            // assert the driver's job is fully done.
+            parent.cancel()
+            withTimeoutOrNull(5_000L) { driver.driverJob.join() }
+            assertTrue(
+                driver.driverJob.isCompleted,
+                "driver.driverJob must be completed after parent.cancel() — observed isCompleted=" +
+                    "${driver.driverJob.isCompleted} active=${driver.driverJob.isActive}",
+            )
+        }
+
+    @Test
+    fun repeatedSessionLifecycleDoesNotLeakThreads() =
+        runBlocking {
+            // Warm up Dispatchers.IO so its worker count has stabilised
+            // before we sample. Otherwise the first session's growth
+            // (creating fresh IO workers from the pool's lazy init)
+            // shows up as a leak.
+            warmUpDispatchersIo()
+            val baseline = liveThreadCount()
+            val sessions = 100
+            for (i in 0 until sessions) {
+                val parent = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+                val (driver, connection, socket) = startDriver(parent)
+                // Keep the session very short — just long enough that
+                // the driver kicked off both read + send loops and may
+                // have done a PTO retransmit.
+                delay(10)
+                driver.close()
+                withTimeoutOrNull(2_000L) { driver.closeTeardownJob?.join() }
+                    ?: error(
+                        "session $i: close teardown coroutine never completed — driver leaked. " +
+                            "Status=${connection.status}",
+                    )
+                parent.cancel()
+                withTimeoutOrNull(2_000L) { driver.driverJob.join() }
+                    ?: error("session $i: driver job did not complete after parent.cancel()")
+                assertEquals(
+                    QuicConnection.Status.CLOSED,
+                    connection.status,
+                    "session $i: connection must be CLOSED after teardown",
+                )
+                assertTrue(
+                    socketIsClosed(socket),
+                    "session $i: UDP socket must be closed",
+                )
+            }
+
+            // Coroutines may still be wrapping up; give Dispatchers.IO
+            // a small drain window so any in-flight worker reuse
+            // settles before we sample.
+            delay(200)
+            val finalCount = liveThreadCount()
+            val growth = finalCount - baseline
+            // Generous band — the IO pool naturally fluctuates a few
+            // workers. A real leak (one persistent thread per session)
+            // would push growth ≥ ~100. 16 is strictly above the
+            // observed JVM noise band and well below leak.
+            assertTrue(
+                growth <= 16,
+                "thread count grew by $growth across $sessions sessions " +
+                    "(baseline=$baseline final=$finalCount). Anything > 16 indicates a leak.",
+            )
+        }
+
+    /** True if a UDP send on the socket throws — i.e. close() has run. */
+    private fun socketIsClosed(socket: UdpSocket): Boolean =
+        try {
+            runBlocking { socket.send(ByteArray(1)) }
+            false
+        } catch (_: Throwable) {
+            true
+        }
+
+    private suspend fun startDriver(parent: CoroutineScope): Triple<QuicConnectionDriver, QuicConnection, UdpSocket> {
+        val socket = UdpSocket.connect("127.0.0.1", blackhole.localPort)
+        val connection =
+            QuicConnection(
+                serverName = "lifecycle.test",
+                config = QuicConnectionConfig(),
+                tlsCertificateValidator = PermissiveCertificateValidator(),
+            )
+        val driver = QuicConnectionDriver(connection, socket, parent)
+        driver.start()
+        return Triple(driver, connection, socket)
+    }
+
+    private suspend fun warmUpDispatchersIo() {
+        // Touch Dispatchers.IO from a few coroutines so the lazy worker
+        // pool has its baseline workers spun up. Without this the
+        // first measurement's "baseline" is artificially low and the
+        // first sessions' creation of IO workers looks like a leak.
+        repeat(4) {
+            kotlinx.coroutines.withContext(Dispatchers.IO) { Thread.sleep(2) }
+        }
+    }
+
+    private fun liveThreadCount(): Int = Thread.getAllStackTraces().keys.size
+}