From bfc983bff78af6b5274b62ffe9324e5c88cfc119 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 7 May 2026 21:33:59 +0000 Subject: [PATCH] feat(quic): retire fully-settled streams to keep tracker bounded under audio-room churn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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/connection/QuicConnection.kt | 116 ++++- .../quic/connection/QuicConnectionDriver.kt | 24 + .../quic/connection/QuicConnectionWriter.kt | 22 +- .../vitorpamplona/quic/stream/QuicStream.kt | 58 +++ .../connection/StreamRetirementSoakTest.kt | 410 ++++++++++++++++++ .../QuicConnectionDriverLifecycleTest.kt | 220 ++++++++++ 6 files changed, 843 insertions(+), 7 deletions(-) create mode 100644 quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetirementSoakTest.kt create mode 100644 quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriverLifecycleTest.kt diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index 0228ae728..9924fbfb9 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -301,15 +301,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 - * added (no removal in the current model), so the two stay in sync as - * long as `getOrCreatePeerStreamLocked` and `openBidi/UniStream` append - * to both. + * `streams.entries.toList()` allocating per drain. Mutated in lockstep + * with [streams] under [streamsLock]: + * - [openBidiStreamLocked] / [openUniStreamLocked] / + * [getOrCreatePeerStreamLocked] append. + * - [retireFullyDoneStreamsLocked] removes entries whose stream has + * flipped [QuicStream.isFullyRetired] = true. The retire pass + * folds the receive-side high-water mark into + * [retiredStreamsRecvBytes] so the writer's MAX_DATA accounting + * in `appendFlowControlUpdates` doesn't regress when a retired + * stream's `receive.contiguousEnd()` drops out of the iteration. + * + * Without retirement, the moq-lite audio-rooms path leaks one + * QuicStream per Opus frame for the lifetime of the session — the + * stream-cliff investigation in + * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md` + * pegs steady-state churn at ~50 streams/sec, so a 3-hour room + * accumulates ~540 000 entries before retirement was wired. */ private val streamsList = mutableListOf() private var nextLocalBidiIndex: Long = 0L private var nextLocalUniIndex: Long = 0L + /** + * Cumulative `receive.contiguousEnd()` from streams that have been + * removed by [retireFullyDoneStreamsLocked]. Folded into the writer's + * `totalRecvAdvanced` accumulator so MAX_DATA continues to advertise + * the *connection-lifetime* receive high-water mark even after the + * contributing streams have been dropped from [streamsList]. Without + * this, retiring N streams that each delivered K bytes would silently + * regress the advertised connection-level credit by N*K, eventually + * starving the peer once `advertisedMaxData` was tripped past. + * + * Caller of any read/write must hold [streamsLock]. + */ + internal var retiredStreamsRecvBytes: Long = 0L + private set + + /** + * Cumulative count of streams ever removed by + * [retireFullyDoneStreamsLocked]. Diagnostic-only; tests that drive + * stream churn use this to assert retirement actually fired (rather + * than relying on `streamsList.size` alone, which can shrink because + * a soak loop happened to drain the same workload it just enqueued). + * + * Caller of any read/write must hold [streamsLock]. + */ + internal var retiredStreamsCount: Long = 0L + private set + /** * 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 - * remove path exists today; if/when one is added it MUST update both. + * (openBidi/UniStream, getOrCreatePeerStreamLocked, + * retireFullyDoneStreamsLocked) update both. */ internal fun streamsListLocked(): List = streamsList + /** + * Walk [streamsList] once and remove every stream whose + * [QuicStream.isFullyRetired] flag is true. Drops the entry from both + * [streamsList] and [streams]. Receive-side high-water marks fold + * into [retiredStreamsRecvBytes] so the writer's connection-level + * MAX_DATA accounting continues to see the lifetime total. + * + * Returns the number of streams removed in this pass. + * + * Caller MUST hold [streamsLock]. The writer drains under + * `streamsLock`, so calling this from the top of `buildApplicationPacket` + * is the natural place — it runs once per send pass, sees the latest + * FIN/RESET ACK state from the parser's just-finished processing, and + * the iteration order matches the writer's per-pass round-robin + * (so the rotation cursor doesn't accidentally point at a hole). + * + * Why retirement is safe even though the QUIC spec requires the peer + * to deliver retransmits if its ACK never reached us: + * - SEND side waits for `finAcked` / `resetAcked`, i.e. the peer has + * already confirmed receipt of our FIN. After that point the peer + * will not re-send anything on the stream. + * - RECEIVE side waits for the parser to have pushed every byte to + * the application's incoming Channel ([ReceiveBuffer.isFullyRead]). + * 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 = pendingDatagrams diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt index 4c8fab2ec..7d1dfffdc 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -64,6 +64,30 @@ class QuicConnectionDriver( private var readJob: Job? = null private var sendJob: Job? = null + /** + * Test-only handle on the driver's [SupervisorJob]. Used by the + * session-lifecycle leak test in + * [com.vitorpamplona.quic.connection.QuicConnectionDriverLifecycleTest] + * to assert that a closed driver reports `isCompleted = true` once + * its read/send loops have unwound — the inverse of "the driver + * leaked a coroutine past close()". + * + * Production code MUST NOT touch this — the driver lifecycle is + * managed end-to-end by [start] / [close]. + */ + internal val driverJob: Job get() = job + + /** + * Test-only handle on the in-flight teardown coroutine. Returns null + * before [close] has been called; once close runs, the returned Job + * lets the test `join()` until teardown is complete (cancel + socket + * close + read/send join). Pre-existing close() returned immediately + * and provided no synchronous "teardown is done" signal — tests had + * to poll `connection.status == CLOSED` and trust that the rest of + * the cleanup eventually settled. + */ + internal val closeTeardownJob: Job? get() = closeJob + /** * Round-5 concurrency #5: close() guard. A second concurrent invocation * (e.g. session close + read-loop death close racing) used to launch a diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 607279620..728bf6967 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/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() // 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 diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt index 05317cb3b..46d1cd888 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/stream/QuicStream.kt @@ -113,6 +113,64 @@ class QuicStream( val isClosed: Boolean get() = send.finSent && receive.finReceived + /** + * True once both directions are *fully* settled and the stream may be + * removed from the connection's tracking lists (see + * `QuicConnection.retireFullyDoneStreamsLocked`). This is strictly + * stronger than [isClosed]: the latter only requires the FIN bits to + * have been observed, not that the peer has acknowledged our FIN / + * RESET (send side) nor that the application has drained the buffered + * receive bytes (receive side). + * + * Lifetime contract: + * - SEND side done: peer has ACK'd our FIN ([SendBuffer.finAcked]) OR + * we ABORTed the stream and the peer ACK'd our RESET_STREAM + * ([resetAcked]). For [Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL] + * there is no send side, so this leg is trivially done. + * - RECEIVE side done: peer FIN'd ([ReceiveBuffer.finReceived]) AND + * every byte has been delivered from the receive buffer to the + * application's incoming Channel ([ReceiveBuffer.isFullyRead]). + * Once both hold, the parser has already invoked [closeIncoming] + * so any application coroutine still draining the buffered + * [incoming] flow will terminate naturally — the QuicStream object + * can outlive its membership in the connection's `streamsList`. + * For [Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE] there is no + * receive side, so this leg is trivially done. + * + * The motivation is the moq-lite audio-rooms path: each Opus frame is + * forwarded as a fresh peer-uni stream by the relay. A 3-hour session + * at ~50 frames/sec churns ~540 000 streams. Without retirement, + * `streamsList` and the `streams` map grow monotonically — the + * `nestsClient/plans/2026-04-26-moq-lite-gap.md` soak target wants + * memory flat past handshake-stable. Stream retirement is the only + * QUIC-level fix that keeps the tracker bounded for that workload. + * + * Read this from any thread under the connection's `streamsLock` — the + * underlying flags are `@Volatile` and the buffers' synchronized + * blocks publish their state atomically. + */ + val isFullyRetired: Boolean + get() { + val sendSettled = + when (direction) { + Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL -> true + + Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE, + Direction.BIDIRECTIONAL, + -> send.finAcked || resetAcked + } + if (!sendSettled) return false + val recvSettled = + when (direction) { + Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE -> true + + Direction.UNIDIRECTIONAL_REMOTE_TO_LOCAL, + Direction.BIDIRECTIONAL, + -> receive.finReceived && receive.isFullyRead() + } + return recvSettled + } + /** * Pushes [data] toward the consumer. Returns false if the bounded channel * was full; the caller (parser) is expected to escalate to a connection- diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetirementSoakTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/StreamRetirementSoakTest.kt new file mode 100644 index 000000000..191c81b6c --- /dev/null +++ b/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 = + 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 + } +} diff --git a/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriverLifecycleTest.kt b/quic/src/jvmTest/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriverLifecycleTest.kt new file mode 100644 index 000000000..4f05a644e --- /dev/null +++ b/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 { + 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 +}