From 96a585a67e0ee8b989ab343f3b91dbb3d559c248 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 1 May 2026 13:59:38 +0000 Subject: [PATCH] Revert "quic: suspend openUni/BidiStream on peer-cap exhaustion + emit STREAMS_BLOCKED" This reverts commit f0705e3ab13472c719373f1eb2334e373ba6b1be. --- .../quic/connection/QuicConnection.kt | 182 +++-------- .../quic/connection/QuicConnectionDriver.kt | 5 - .../quic/connection/QuicConnectionParser.kt | 16 - .../quic/connection/QuicConnectionWriter.kt | 14 - .../com/vitorpamplona/quic/frame/Frame.kt | 29 +- .../quic/connection/PeerStreamLimitTest.kt | 293 ++---------------- 6 files changed, 63 insertions(+), 476 deletions(-) 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 2d3afee8d..5b9fcc4f4 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -193,58 +193,6 @@ class QuicConnection( */ private val closedSignal = Channel(Channel.CONFLATED) - /** - * Notifier completed every time the peer extends our stream cap (either - * direction) via a MAX_STREAMS frame, OR when the connection closes. - * Suspended [openBidiStream] / [openUniStream] callers await the current - * notifier; once it fires, they re-acquire the lock and re-check whether - * they have credit. The notifier is replaced after each fire so a fresh - * await reads the next cycle. - * - * Only mutated under [lock]. Reading the reference is safe without the - * lock IF the caller subsequently takes the lock and re-checks the cap — - * the read might be stale, the await might be stale, but the loop - * guarantees forward progress because the parser also signals after each - * cap raise. - */ - private var streamCapNotifier: kotlinx.coroutines.CompletableDeferred = - kotlinx.coroutines.CompletableDeferred() - - /** - * Stream-limit values we owe the peer in STREAMS_BLOCKED frames. Set - * when [openBidiStream] / [openUniStream] discover the cap is exhausted. - * Cleared by [QuicConnectionWriter] after the frame is emitted. Per RFC - * 9000 §19.14 we should send STREAMS_BLOCKED at most once per cap value - * — null means "nothing to send". Reset to a new value if the cap - * advances and we hit it again later. - */ - internal var pendingStreamsBlockedBidi: Long? = null - internal var pendingStreamsBlockedUni: Long? = null - - /** - * Hook invoked when the connection wants to nudge the writer (e.g. a - * suspended opener queued STREAMS_BLOCKED and needs the writer to - * flush). Wired by [QuicConnectionDriver.start] and reset on close. - * Null while the driver is not running (in-process tests etc) — the - * connection still works, just without an external send loop. - */ - @Volatile - internal var sendWakeupHook: (() -> Unit)? = null - - /** - * Wake any suspended [openBidiStream] / [openUniStream] callers. Called - * by the parser after each MAX_STREAMS frame raises a cap, and by the - * close path so blocked openers can throw [QuicConnectionClosedException] - * instead of suspending forever. - * - * Caller must hold [lock] (the notifier ref is mutated here). - */ - internal fun signalStreamCapLocked() { - val old = streamCapNotifier - streamCapNotifier = kotlinx.coroutines.CompletableDeferred() - old.complete(Unit) - } - private val tlsListener = object : TlsSecretsListener { override fun onHandshakeKeysReady( @@ -390,98 +338,47 @@ class QuicConnection( val lock: Mutex = Mutex() /** - * Allocate a new client-initiated bidirectional stream. + * Allocate a new client-initiated bidirectional stream. Locked. * - * If the peer has not granted enough bidi stream credit yet, this - * method **suspends** (not throws) until either: - * - an inbound MAX_STREAMS frame raises the cap (RFC 9000 §19.11), - * or - * - the connection closes — in which case - * [QuicConnectionClosedException] is thrown. - * - * Before suspending, queues a STREAMS_BLOCKED_BIDI frame (RFC 9000 - * §19.14) so the peer knows we want more credit. Without this, a - * conservative peer that only extends MAX_STREAMS in response to - * STREAMS_BLOCKED would never grant new credit, deadlocking the - * client. + * Throws [QuicStreamLimitException] if the peer has not granted enough + * bidirectional stream credit yet. Use [peerMaxStreamsBidiSnapshot] to + * check capacity proactively if the caller wants to back-pressure rather + * than throw. */ - suspend fun openBidiStream(): QuicStream = openClientStream(uni = false) - - /** - * Allocate a new client-initiated unidirectional (write-only) stream. - * Suspends if the peer has exhausted our uni stream cap; see - * [openBidiStream] for the semantics. - */ - suspend fun openUniStream(): QuicStream = openClientStream(uni = true) - - /** - * Shared body for [openBidiStream] / [openUniStream]. The two paths - * are byte-for-byte symmetric except for the StreamId kind, the per- - * direction caps + indices, the per-direction `sendCredit` source, - * and the receive limit (uni-out streams cannot receive). - * - * Re-acquires the lock on every retry — required to read the current - * `peerMaxStreams*` (mutated by the parser under the lock) and to - * mutate `nextLocal*Index` atomically with the streams map. - */ - private suspend fun openClientStream(uni: Boolean): QuicStream { - while (true) { - // Capture under lock: either we have credit and allocate, or - // we record our blocked-at limit and grab a notifier ref to - // await *outside* the lock (otherwise the parser couldn't - // acquire the lock to raise the cap, deadlocking). - val notifier: kotlinx.coroutines.CompletableDeferred - lock.withLock { - if (status == Status.CLOSED) { - throw QuicConnectionClosedException( - "connection closed before stream could be allocated", - ) - } - val nextIndex = if (uni) nextLocalUniIndex else nextLocalBidiIndex - val cap = if (uni) peerMaxStreamsUni else peerMaxStreamsBidi - if (nextIndex < cap) { - val id = - StreamId.build( - if (uni) StreamId.Kind.CLIENT_UNI else StreamId.Kind.CLIENT_BIDI, - nextIndex, - ) - if (uni) nextLocalUniIndex++ else nextLocalBidiIndex++ - val stream = - QuicStream( - id, - if (uni) QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE else QuicStream.Direction.BIDIRECTIONAL, - ) - stream.sendCredit = - if (uni) { - peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni - } else { - peerTransportParameters?.initialMaxStreamDataBidiRemote ?: config.initialMaxStreamDataBidiRemote - } - stream.receiveLimit = if (uni) 0L else config.initialMaxStreamDataBidiLocal - streams[id] = stream - streamsList += stream - return stream - } - // Out of credit. Queue a STREAMS_BLOCKED frame at the - // current cap (per RFC 9000 §19.14, "stream limit" = - // count from MAX_STREAMS) so the peer knows we want more. - // Replace any earlier pending entry — we only ever owe - // the peer one STREAMS_BLOCKED per cap value. - if (uni) { - pendingStreamsBlockedUni = cap - } else { - pendingStreamsBlockedBidi = cap - } - notifier = streamCapNotifier + suspend fun openBidiStream(): QuicStream = + lock.withLock { + if (nextLocalBidiIndex >= peerMaxStreamsBidi) { + throw QuicStreamLimitException( + "peer-granted bidi stream cap reached " + + "(used=$nextLocalBidiIndex limit=$peerMaxStreamsBidi)", + ) } - // Nudge the writer so STREAMS_BLOCKED actually leaves the - // host. Without this, a quiescent connection (no other - // writes pending) would hold the frame in the writer's - // queue until the next unrelated wakeup. - sendWakeupHook?.invoke() - notifier.await() + val id = StreamId.build(StreamId.Kind.CLIENT_BIDI, nextLocalBidiIndex++) + val stream = QuicStream(id, QuicStream.Direction.BIDIRECTIONAL) + stream.sendCredit = peerTransportParameters?.initialMaxStreamDataBidiRemote ?: config.initialMaxStreamDataBidiRemote + stream.receiveLimit = config.initialMaxStreamDataBidiLocal + streams[id] = stream + streamsList += stream + stream + } + + /** Allocate a new client-initiated unidirectional (write-only) stream. Locked. */ + suspend fun openUniStream(): QuicStream = + lock.withLock { + if (nextLocalUniIndex >= peerMaxStreamsUni) { + throw QuicStreamLimitException( + "peer-granted uni stream cap reached " + + "(used=$nextLocalUniIndex limit=$peerMaxStreamsUni)", + ) + } + val id = StreamId.build(StreamId.Kind.CLIENT_UNI, nextLocalUniIndex++) + val stream = QuicStream(id, QuicStream.Direction.UNIDIRECTIONAL_LOCAL_TO_REMOTE) + stream.sendCredit = peerTransportParameters?.initialMaxStreamDataUni ?: config.initialMaxStreamDataUni + stream.receiveLimit = 0L // can't receive + streams[id] = stream + streamsList += stream + stream } - } /** Snapshot of peer-granted bidi cap. Reads do not need the lock — long writes are atomic on every supported platform. */ fun peerMaxStreamsBidiSnapshot(): Long = peerMaxStreamsBidi @@ -588,11 +485,6 @@ class QuicConnection( closedSignal.close() peerStreamSignal.close() incomingDatagramSignal.close() - // Wake any openBidiStream / openUniStream callers blocked on the - // stream-cap notifier so they re-acquire the lock, observe - // status == CLOSED, and throw QuicConnectionClosedException - // instead of hanging forever. - if (!streamCapNotifier.isCompleted) streamCapNotifier.complete(Unit) } /** 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 00f2933b0..e3baf181d 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -73,11 +73,6 @@ class QuicConnectionDriver( fun start() { connection.start() - // Wire the connection's sendWakeupHook so internal events - // (a suspended openUniStream queueing STREAMS_BLOCKED, the close - // path completing handshakeDoneSignal, etc) can nudge the send - // loop without callers having to know about the driver. - connection.sendWakeupHook = { sendWakeup.trySend(Unit) } readJob = scope.launch { readLoop() } sendJob = scope.launch { sendLoop() } // Initial nudge so the ClientHello goes out immediately. diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt index adc040266..abd87f92b 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionParser.kt @@ -35,7 +35,6 @@ import com.vitorpamplona.quic.frame.PingFrame import com.vitorpamplona.quic.frame.ResetStreamFrame import com.vitorpamplona.quic.frame.StopSendingFrame import com.vitorpamplona.quic.frame.StreamFrame -import com.vitorpamplona.quic.frame.StreamsBlockedFrame import com.vitorpamplona.quic.frame.decodeFrames import com.vitorpamplona.quic.packet.LongHeaderPacket import com.vitorpamplona.quic.packet.LongHeaderType @@ -287,30 +286,15 @@ private fun dispatchFrames( // RFC 9000 §19.11: MAX_STREAMS only ever raises the cap. // Frames with values smaller than the current cap are ignored. // Bidi vs uni is signaled via the frame's `bidi` flag. - var raised = false if (frame.bidi) { if (frame.maxStreams > conn.peerMaxStreamsBidi) { conn.peerMaxStreamsBidi = frame.maxStreams - raised = true } } else { if (frame.maxStreams > conn.peerMaxStreamsUni) { conn.peerMaxStreamsUni = frame.maxStreams - raised = true } } - // Wake any [openBidiStream] / [openUniStream] callers blocked - // on the previous cap so they re-evaluate. Caller (read loop) - // already holds [conn.lock], so the notifier swap is safe. - if (raised) conn.signalStreamCapLocked() - } - - is StreamsBlockedFrame -> { - // We don't currently advertise dynamic stream caps to the - // peer beyond the initial transport parameters, so the - // peer's STREAMS_BLOCKED is informational only. The frame - // is ack-eliciting per RFC 9000 §13.2.1. - ackEliciting = true } is ResetStreamFrame -> { 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 6a536b6bb..3924d90a6 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -27,7 +27,6 @@ import com.vitorpamplona.quic.frame.Frame import com.vitorpamplona.quic.frame.MaxDataFrame import com.vitorpamplona.quic.frame.MaxStreamDataFrame import com.vitorpamplona.quic.frame.StreamFrame -import com.vitorpamplona.quic.frame.StreamsBlockedFrame import com.vitorpamplona.quic.frame.encodeFrames import com.vitorpamplona.quic.packet.LongHeaderPacket import com.vitorpamplona.quic.packet.LongHeaderPlaintextPacket @@ -251,19 +250,6 @@ private fun buildApplicationPacket( // stream and MAX_DATA at the connection level. appendFlowControlUpdates(conn, frames) - // RFC 9000 §19.14: tell the peer we're starving for stream IDs. - // Drained on each emission so we send at most one STREAMS_BLOCKED per - // cap value per direction (re-set by [openClientStream] only if we - // hit the cap again with a higher value). - conn.pendingStreamsBlockedBidi?.let { limit -> - frames += StreamsBlockedFrame(bidi = true, streamLimit = limit) - conn.pendingStreamsBlockedBidi = null - } - conn.pendingStreamsBlockedUni?.let { limit -> - frames += StreamsBlockedFrame(bidi = false, streamLimit = limit) - conn.pendingStreamsBlockedUni = null - } - // Pending datagrams while (conn.pendingDatagramsLocked().isNotEmpty()) { val payload = conn.pendingDatagramsLocked().removeFirst() diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/frame/Frame.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/frame/Frame.kt index a7b6997e7..e2c47f379 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/frame/Frame.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/frame/Frame.kt @@ -242,27 +242,6 @@ class MaxStreamsFrame( } } -/** - * RFC 9000 §19.14. Sent by an endpoint that wants to open a stream of the - * given direction but has exhausted its peer-granted cap. Carries the - * highest stream id the sender currently believes it is allowed to open - * (i.e. the count it received in the last MAX_STREAMS frame for that - * direction). - * - * Required for proper flow-control bookkeeping — without it, the peer - * has no signal that we're starved and may delay extending credit. The - * frame itself is informational; it doesn't mutate the cap. - */ -class StreamsBlockedFrame( - val bidi: Boolean, - val streamLimit: Long, -) : Frame() { - override fun encode(out: QuicWriter) { - out.writeByte(if (bidi) FrameType.STREAMS_BLOCKED_BIDI.toInt() else FrameType.STREAMS_BLOCKED_UNI.toInt()) - out.writeVarint(streamLimit) - } -} - class DatagramFrame( val data: ByteArray, val explicitLength: Boolean = true, @@ -412,12 +391,8 @@ fun decodeFrames(data: ByteArray): List { r.readVarint() } - type == FrameType.STREAMS_BLOCKED_BIDI -> { - out += StreamsBlockedFrame(bidi = true, streamLimit = r.readVarint()) - } - - type == FrameType.STREAMS_BLOCKED_UNI -> { - out += StreamsBlockedFrame(bidi = false, streamLimit = r.readVarint()) + type == FrameType.STREAMS_BLOCKED_BIDI || type == FrameType.STREAMS_BLOCKED_UNI -> { + r.readVarint() } type == FrameType.NEW_CONNECTION_ID -> { diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamLimitTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamLimitTest.kt index 3296df65f..651545fa3 100644 --- a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamLimitTest.kt +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamLimitTest.kt @@ -21,46 +21,29 @@ package com.vitorpamplona.quic.connection import com.vitorpamplona.quic.frame.MaxStreamsFrame -import com.vitorpamplona.quic.frame.StreamsBlockedFrame -import com.vitorpamplona.quic.frame.decodeFrames import com.vitorpamplona.quic.tls.InProcessTlsServer -import kotlinx.coroutines.CompletableDeferred -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.async -import kotlinx.coroutines.cancelAndJoin -import kotlinx.coroutines.delay -import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking -import kotlinx.coroutines.withTimeoutOrNull import kotlin.test.Test import kotlin.test.assertEquals -import kotlin.test.assertNotNull -import kotlin.test.assertNull +import kotlin.test.assertFailsWith import kotlin.test.assertTrue /** - * Verifies the per-direction stream-cap flow-control behaviour: + * Verifies the audit-3 fix: peer-granted stream concurrency limits are now + * tracked and enforced. * - * - Peer transport parameters at handshake (`initial_max_streams_bidi/uni`) - * bound how many client-initiated streams [QuicConnection.openBidiStream] - * / [QuicConnection.openUniStream] may allocate before suspending. - * - Subsequent inbound MAX_STREAMS frames (RFC 9000 §19.11) raise the cap - * and wake suspended openers. - * - When an opener suspends, a STREAMS_BLOCKED frame (RFC 9000 §19.14) - * is queued so the peer knows we want more credit. - * - Closing the connection wakes any blocked opener with a - * [QuicConnectionClosedException] rather than hanging it. + * Two paths feed the cap: + * 1. Peer transport parameters at handshake (`initial_max_streams_bidi/uni`) + * 2. Subsequent MAX_STREAMS frames (RFC 9000 §19.11) * - * The previous behaviour — throwing [QuicStreamLimitException] synchronously - * when the cap was exhausted — pushed the back-pressure problem onto every - * caller (which usually swallowed it via `runCatching` and silently dropped - * data). See nestsClient sweep results for the symptom. + * Without this enforcement, [QuicConnection.openBidiStream] silently allocated + * stream IDs past the cap, and the peer eventually closed the connection with + * STREAM_LIMIT_ERROR — a failure that surfaced as "connection randomly drops + * after a burst of opens" rather than a clean error. */ class PeerStreamLimitTest { @Test - fun open_bidi_suspends_when_peer_advertises_zero_bidi_streams() { + fun open_bidi_throws_when_peer_advertises_zero_bidi_streams() { runBlocking { val client = QuicConnection( @@ -70,6 +53,8 @@ class PeerStreamLimitTest { com.vitorpamplona.quic.tls .PermissiveCertificateValidator(), ) + // Explicitly advertise zero bidi streams. (The pipe's default TPs + // grant 16, so we override.) val serverScid = ConnectionId.random(8) val tlsServer = InProcessTlsServer( @@ -96,18 +81,14 @@ class PeerStreamLimitTest { pipe.drive(maxRounds = 16) assertEquals(QuicConnection.Status.CONNECTED, client.status) - // openBidiStream now SUSPENDS instead of throwing — the peer - // hasn't granted any credit, so we should never resolve. - val opened = - withTimeoutOrNull(150L) { - client.openBidiStream() - } - assertNull(opened, "openBidiStream must suspend when peer cap = 0; got $opened") + assertFailsWith { + client.openBidiStream() + } } } @Test - fun open_bidi_succeeds_within_peer_cap_then_suspends_at_boundary() { + fun open_bidi_succeeds_within_peer_advertised_cap_and_throws_at_boundary() { runBlocking { val client = QuicConnection( @@ -117,6 +98,8 @@ class PeerStreamLimitTest { com.vitorpamplona.quic.tls .PermissiveCertificateValidator(), ) + // Build the TLS server with the audit-4 #7 required CIDs plus + // tight stream caps for the boundary test. val serverScid = ConnectionId.random(8) val serverTpBytes = TransportParameters( @@ -142,226 +125,15 @@ class PeerStreamLimitTest { assertEquals(QuicConnection.Status.CONNECTED, client.status) assertEquals(3L, client.peerMaxStreamsBidiSnapshot()) - // Three opens within the cap should resolve instantly. + // Three opens should succeed. client.openBidiStream() client.openBidiStream() client.openBidiStream() - // Fourth must SUSPEND (not throw) — we'd otherwise violate the - // peer's cap and trigger STREAM_LIMIT_ERROR on their side. - val fourth = - withTimeoutOrNull(150L) { - client.openBidiStream() - } - assertNull(fourth, "fourth openBidiStream must suspend; got $fourth") - } - } - - @Test - fun max_streams_uni_frame_wakes_suspended_open_uni_stream() { - runBlocking { - val client = - QuicConnection( - serverName = "example.test", - config = QuicConnectionConfig(), - tlsCertificateValidator = - com.vitorpamplona.quic.tls - .PermissiveCertificateValidator(), - ) - val serverScid = ConnectionId.random(8) - val serverTpBytes = - TransportParameters( - initialMaxData = 1_000_000, - initialMaxStreamDataBidiLocal = 100_000, - initialMaxStreamDataBidiRemote = 100_000, - initialMaxStreamDataUni = 100_000, - // Peer initially grants ZERO uni streams — opener will - // suspend until a MAX_STREAMS_UNI frame raises the cap. - initialMaxStreamsBidi = 100, - initialMaxStreamsUni = 0, - initialSourceConnectionId = serverScid.bytes, - originalDestinationConnectionId = client.destinationConnectionId.bytes, - ).encode() - val tlsServer = InProcessTlsServer(transportParameters = serverTpBytes) - 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) - assertEquals(0L, client.peerMaxStreamsUniSnapshot()) - - // Launch a coroutine that tries to open a uni stream. It must - // suspend immediately because the cap is 0. - val supervisor = SupervisorJob() - val scope = CoroutineScope(supervisor + Dispatchers.Default) - val openResult = CompletableDeferred() - val opener = - scope.launch { - val s = client.openUniStream() - openResult.complete(s) - } - // Give the launcher a moment to actually park on the notifier. - delay(50L) - assertTrue(opener.isActive, "opener should still be suspended pending credit") - - // Now feed a MAX_STREAMS_UNI(2) frame and process it as if it - // had arrived from the peer. The parser raises peerMaxStreamsUni - // and signals the cap notifier; the suspended opener wakes up - // and allocates stream id 2 (CLIENT_UNI #0). - // The MAX_STREAMS_UNI(2) frame on its own is only ~2 bytes; - // QUIC packets need ≥4 bytes of protected payload after the - // packet number for the HP sample (RFC 9001 §5.4.2). Pad - // with a PING (1 byte) and a few PaddingFrames-equivalents. - val datagram = - pipe.buildServerApplicationDatagram( - listOf( - com.vitorpamplona.quic.frame.PingFrame, - com.vitorpamplona.quic.frame.PingFrame, - com.vitorpamplona.quic.frame.PingFrame, - com.vitorpamplona.quic.frame.PingFrame, - MaxStreamsFrame(bidi = false, maxStreams = 2), - ), - ) ?: error("server has no application keys yet") - feedDatagram(client, datagram, nowMillis = 0L) - - val opened = - withTimeoutOrNull(500L) { openResult.await() } - assertNotNull(opened, "opener should have resumed after MAX_STREAMS_UNI raised cap") - assertEquals(2L, opened.streamId, "first client uni stream id is 2 (uni-low encoding)") - opener.cancelAndJoin() - supervisor.cancelAndJoin() - } - } - - @Test - fun open_uni_stream_queues_streams_blocked_frame_when_cap_is_zero() { - runBlocking { - val client = - QuicConnection( - serverName = "example.test", - config = QuicConnectionConfig(), - tlsCertificateValidator = - com.vitorpamplona.quic.tls - .PermissiveCertificateValidator(), - ) - val serverScid = ConnectionId.random(8) - val tlsServer = - InProcessTlsServer( - transportParameters = - TransportParameters( - initialMaxData = 1_000_000, - initialMaxStreamDataBidiLocal = 100_000, - initialMaxStreamDataBidiRemote = 100_000, - initialMaxStreamDataUni = 100_000, - initialMaxStreamsBidi = 100, - initialMaxStreamsUni = 0, - 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) - - val supervisor = SupervisorJob() - val scope = CoroutineScope(supervisor + Dispatchers.Default) - scope.launch { client.openUniStream() } - delay(50L) - - // Opener should have recorded the cap value it hit so the - // writer emits a STREAMS_BLOCKED_UNI on the next drain. - assertEquals( - 0L, - client.pendingStreamsBlockedUni, - "openUniStream must queue STREAMS_BLOCKED_UNI(0) when peer cap = 0", - ) - - // Drain rounds eventually clear the slot once the writer - // emits the frame. We don't decode the wire bytes here (a - // single STREAMS_BLOCKED frame is too small for the QUIC HP - // sample on its own); instead we trust the writer integration - // covered by [streams_blocked_frame_roundtrips_via_decode_frames] - // and verify the queue-and-clear bookkeeping at the - // connection level. - // Drain may produce no packet under HP-sample padding rules, - // but the field-clear behaviour is exercised by the writer - // having access to and zeroing the slot once the frame is - // appended to the outbound list. [drainOutbound] returns null - // when the resulting packet is too small for HP sampling, but - // by that point the writer has already moved the value out of - // pendingStreamsBlockedUni into the local frames list. - - supervisor.cancelAndJoin() - } - } - - @Test - fun closing_connection_wakes_blocked_open_with_closed_exception() { - runBlocking { - val client = - QuicConnection( - serverName = "example.test", - config = QuicConnectionConfig(), - tlsCertificateValidator = - com.vitorpamplona.quic.tls - .PermissiveCertificateValidator(), - ) - val serverScid = ConnectionId.random(8) - val tlsServer = - InProcessTlsServer( - transportParameters = - TransportParameters( - initialMaxData = 1_000_000, - initialMaxStreamDataBidiLocal = 100_000, - initialMaxStreamDataBidiRemote = 100_000, - initialMaxStreamDataUni = 100_000, - initialMaxStreamsBidi = 0, - initialMaxStreamsUni = 0, - 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) - - val supervisor = SupervisorJob() - val scope = CoroutineScope(supervisor + Dispatchers.Default) - val openCall = - scope.async { - runCatching { client.openBidiStream() } - } - delay(50L) - assertTrue(openCall.isActive, "opener must be suspended pending credit") - - client.markClosedExternally("test") - val r = - withTimeoutOrNull(500L) { openCall.await() } - assertNotNull(r, "opener should have resumed once connection closed") - assertTrue( - r.exceptionOrNull() is QuicConnectionClosedException, - "closed connection must throw QuicConnectionClosedException, got ${r.exceptionOrNull()}", - ) - supervisor.cancelAndJoin() + // Fourth must throw — we'd otherwise violate the peer's cap. + assertFailsWith { + client.openBidiStream() + } } } @@ -382,21 +154,4 @@ class PeerStreamLimitTest { assertTrue(frame.bidi) assertEquals(100L, frame.maxStreams) } - - @Test - fun streams_blocked_frame_roundtrips_via_decode_frames() { - val encoded = - com.vitorpamplona.quic.frame.encodeFrames( - listOf( - StreamsBlockedFrame(bidi = true, streamLimit = 7), - StreamsBlockedFrame(bidi = false, streamLimit = 99), - ), - ) - val decoded = decodeFrames(encoded).filterIsInstance() - assertEquals(2, decoded.size) - assertEquals(true, decoded[0].bidi) - assertEquals(7L, decoded[0].streamLimit) - assertEquals(false, decoded[1].bidi) - assertEquals(99L, decoded[1].streamLimit) - } }