test(quic): de-flake PeerUniStreamDrainTest slow-consumer overflow

drainPeerInitiatedUniStreamsIntoBlackHole_keeps_connection_alive passed
in isolation but flaked under parallel load with the audit-4 #3
"slow consumer overflowed incoming channel" teardown.

Root cause: the test ran the black-hole drainer on the shared
Dispatchers.Default pool. The drainer only has to keep the 64-deep
per-stream incomingChannel drained, but when the rest of the JVM test
suite saturates every Default worker thread the drainer coroutine can be
denied a scheduling slot long enough for the synchronous feed loop to
push more than 64 chunks ahead of it, tripping the slow-consumer
teardown. The fixed-delay yields (delay(1) every 16 chunks) do not help
when all pool threads are pinned.

Fix: run the drainer on the runBlocking coroutine's own single-threaded
event loop (CoroutineScope(coroutineContext + SupervisorJob())) and pace
the feed loop with cooperative yield() instead of wall-clock delays. Each
yield deterministically runs the drainer's queued channel-receive
continuation before the producer resumes, so the buffer never approaches
its bound regardless of machine load. The drain contract under test
(parser trySend -> incomingChannel -> collect -> discard) is exercised
identically; only the scheduling is pinned.

Verified with a temporary reproduction that pins every Default worker
thread: the old Default-scheduled drainer overflows to CLOSED while the
confined drainer stays CONNECTED under the same saturation.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MGZt6D298kGv2XgAi3Uuaj
This commit is contained in:
Claude
2026-07-29 15:02:16 +00:00
parent 3cfa2e0efb
commit 21c44cfc36
@@ -24,12 +24,10 @@ import com.vitorpamplona.quic.frame.StreamFrame
import com.vitorpamplona.quic.stream.StreamId import com.vitorpamplona.quic.stream.StreamId
import com.vitorpamplona.quic.tls.InProcessTlsServer import com.vitorpamplona.quic.tls.InProcessTlsServer
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel import kotlinx.coroutines.cancel
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout import kotlinx.coroutines.yield
import kotlin.test.Test import kotlin.test.Test
import kotlin.test.assertEquals import kotlin.test.assertEquals
import kotlin.test.assertNotEquals import kotlin.test.assertNotEquals
@@ -123,8 +121,30 @@ class PeerUniStreamDrainTest {
pipe.drive(maxRounds = 16) pipe.drive(maxRounds = 16)
assertEquals(QuicConnection.Status.CONNECTED, client.status) assertEquals(QuicConnection.Status.CONNECTED, client.status)
// Wire the explicit drainer BEFORE pushing the bytes. // Run the drainer on THIS runBlocking coroutine's own
val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) // single-threaded event loop rather than the shared
// Dispatchers.Default pool.
//
// Why this matters for flakiness: the black-hole drainer only has
// to keep the 64-deep per-stream `incomingChannel` from filling.
// On Dispatchers.Default that is a race against the pool — when the
// rest of the JVM test suite saturates every Default worker thread,
// the drainer coroutine can be denied a scheduling slot long enough
// for the synchronous feed loop below to push more than 64 chunks
// ahead of it. That trips the audit-4 #3 slow-consumer teardown and
// the test fails under parallel load while passing in isolation —
// classic shared-pool starvation flakiness.
//
// Confining the drainer to the producer's event loop makes the
// hand-off deterministic: each cooperative `yield()` below parks the
// feed loop and lets the drainer empty the channel before more
// chunks are pushed, so the buffer never approaches its bound
// regardless of machine load. The drain contract under test (parser
// `trySend` -> `incomingChannel` -> collect -> discard) is exercised
// identically; only the scheduling is pinned. The private
// SupervisorJob keeps `scope.cancel()` in the finally block from
// ever touching the runBlocking job itself.
val scope = CoroutineScope(coroutineContext + SupervisorJob())
try { try {
client.drainPeerInitiatedUniStreamsIntoBlackHole(scope) client.drainPeerInitiatedUniStreamsIntoBlackHole(scope)
@@ -142,16 +162,16 @@ class PeerUniStreamDrainTest {
offset += chunk.size.toLong() offset += chunk.size.toLong()
val packet = pipe.buildServerApplicationDatagram(listOf(frame))!! val packet = pipe.buildServerApplicationDatagram(listOf(frame))!!
feedDatagram(client, packet, nowMillis = 0L) feedDatagram(client, packet, nowMillis = 0L)
// Yield occasionally so the drainer coroutine actually // Hand the confined drainer a turn every few chunks so it
// gets a chance to consume — feedDatagram is synchronous // drains the buffer well before it can reach the 64-chunk
// and the drainer launched with Dispatchers.Default needs // bound. feedDatagram is synchronous; yield() on the shared
// a scheduling tick. // event loop deterministically runs the drainer's queued
if (i % 16 == 15) { // channel-receive continuation before the producer resumes.
withTimeout(2_000) { delay(1) } if (i % 16 == 15) yield()
} }
} // Let the drainer consume anything still buffered before we
// Ensure the drainer has caught up before we sample status. // sample status.
withTimeout(2_000) { delay(50) } yield()
assertEquals( assertEquals(
QuicConnection.Status.CONNECTED, QuicConnection.Status.CONNECTED,