Revert "quic: suspend openUni/BidiStream on peer-cap exhaustion + emit STREAMS_BLOCKED"
This reverts commit f0705e3ab1.
This commit is contained in:
Claude
@@ -193,58 +193,6 @@ class QuicConnection(
|
||||
*/
|
||||
private val closedSignal = Channel<Unit>(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<Unit> =
|
||||
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<Unit>
|
||||
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)
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 -> {
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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<Frame> {
|
||||
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 -> {
|
||||
|
||||
+24
-269
@@ -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<QuicStreamLimitException> {
|
||||
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<com.vitorpamplona.quic.stream.QuicStream>()
|
||||
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<QuicStreamLimitException> {
|
||||
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<StreamsBlockedFrame>()
|
||||
assertEquals(2, decoded.size)
|
||||
assertEquals(true, decoded[0].bidi)
|
||||
assertEquals(7L, decoded[0].streamLimit)
|
||||
assertEquals(false, decoded[1].bidi)
|
||||
assertEquals(99L, decoded[1].streamLimit)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user