diff --git a/h3/src/client/builder.rs b/h3/src/client/builder.rs index 5ae628d7..7d8982c4 100644 --- a/h3/src/client/builder.rs +++ b/h3/src/client/builder.rs @@ -78,6 +78,21 @@ impl Builder { self } + /// Limits accepted extension streams buffered while waiting to be claimed by a consumer. + /// + /// The limit is shared across the connection and currently applies to WebTransport + /// unidirectional streams. Claimed streams no longer count towards this limit. + /// Excess streams are rejected using the extension's error code. + /// + /// Defaults to `usize::MAX`, effectively unlimited. Configure a finite limit when + /// enabling extensions that buffer incoming streams. A value of zero disables + /// buffering. This is a local resource limit and is not advertised to the peer + /// in HTTP/3 SETTINGS. + pub fn max_buffered_extension_streams(&mut self, value: usize) -> &mut Self { + self.config.max_buffered_extension_streams = value; + self + } + /// Set the maximum header size this client is willing to accept /// /// See [header size constraints] section of the specification for details. diff --git a/h3/src/config.rs b/h3/src/config.rs index c2a6fe64..8ddcd4c4 100644 --- a/h3/src/config.rs +++ b/h3/src/config.rs @@ -17,6 +17,9 @@ pub struct Config { /// HTTP/3 Settings pub settings: Settings, + + /// Maximum number of extension streams buffered awaiting a consumer. + pub(crate) max_buffered_extension_streams: usize, } /// HTTP/3 Settings @@ -76,6 +79,7 @@ impl TryFrom for frame::Settings { let Config { send_grease, + max_buffered_extension_streams: _, #[cfg(test)] send_settings: _, settings: @@ -171,6 +175,7 @@ impl Default for Config { fn default() -> Self { Self { send_grease: true, + max_buffered_extension_streams: usize::MAX, #[cfg(test)] send_settings: true, settings: Default::default(), diff --git a/h3/src/connection.rs b/h3/src/connection.rs index 44302f1a..18b9ef83 100644 --- a/h3/src/connection.rs +++ b/h3/src/connection.rs @@ -460,12 +460,18 @@ where ))); } } - AcceptedRecvStream::WebTransportUni(id, s) + AcceptedRecvStream::WebTransportUni(id, mut s) if self.config.settings.enable_webtransport => { // Store until someone else picks it up, like a webtransport session which is // not yet established. - self.accepted_streams.wt_uni_streams.push((id, s)) + if self.accepted_streams.wt_uni_streams.len() + >= self.config.max_buffered_extension_streams + { + s.stop_sending(Code::H3_WEBTRANSPORT_BUFFERED_STREAM_REJECTED.into()); + } else { + self.accepted_streams.wt_uni_streams.push((id, s)); + } } //= https://www.rfc-editor.org/rfc/rfc9114#section-6.2.3 diff --git a/h3/src/error/codes.rs b/h3/src/error/codes.rs index cdcbac1c..28b6b4cc 100644 --- a/h3/src/error/codes.rs +++ b/h3/src/error/codes.rs @@ -67,6 +67,10 @@ macro_rules! codes { } codes! { + /// A WebTransport stream was rejected because the extension stream buffer is full. + /// See: + (0x3994bd84, H3_WEBTRANSPORT_BUFFERED_STREAM_REJECTED); + /// Datagram or capsule parse error /// See: (0x33, H3_DATAGRAM_ERROR); diff --git a/h3/src/server/builder.rs b/h3/src/server/builder.rs index 08be4f47..0f00db2a 100644 --- a/h3/src/server/builder.rs +++ b/h3/src/server/builder.rs @@ -66,6 +66,21 @@ impl Builder { self } + /// Limits accepted extension streams buffered while waiting to be claimed by a consumer. + /// + /// The limit is shared across the connection and currently applies to WebTransport + /// unidirectional streams. Claimed streams no longer count towards this limit. + /// Excess streams are rejected using the extension's error code. + /// + /// Defaults to `usize::MAX`, effectively unlimited. Configure a finite limit when + /// enabling extensions that buffer incoming streams. A value of zero disables + /// buffering. This is a local resource limit and is not advertised to the peer + /// in HTTP/3 SETTINGS. + pub fn max_buffered_extension_streams(&mut self, value: usize) -> &mut Self { + self.config.max_buffered_extension_streams = value; + self + } + /// Set the maximum header size this client is willing to accept /// /// See [header size constraints] section of the specification for details. @@ -92,6 +107,9 @@ impl Builder { /// **Server**: /// Supporting for webtransport also requires setting `enable_extended_connect` `enable_datagram` /// and `max_webtransport_sessions`. + /// + /// Also configure [`max_buffered_extension_streams`](Self::max_buffered_extension_streams) + /// to bound streams awaiting a session. The default buffer limit is effectively unlimited. #[inline] pub fn enable_webtransport(&mut self, value: bool) -> &mut Self { self.config.settings.enable_webtransport = value; diff --git a/h3/src/tests/connection.rs b/h3/src/tests/connection.rs index 9f7fb6ef..371943b0 100644 --- a/h3/src/tests/connection.rs +++ b/h3/src/tests/connection.rs @@ -1020,3 +1020,117 @@ async fn h3_quinn_finish_flushes_buffered_data() { // DATA frame: type 0x00, length 7, then the payload. assert_eq!(received, b"\x00\x07payload"); } + +#[tokio::test] +async fn buffered_extension_stream_limit() { + check_buffered_extension_stream_limit(2).await; +} + +#[tokio::test] +async fn buffered_extension_stream_limit_zero() { + check_buffered_extension_stream_limit(0).await; +} + +#[test] +fn buffered_extension_stream_limit_default() { + assert_eq!( + crate::config::Config::default().max_buffered_extension_streams, + usize::MAX + ); +} + +async fn check_buffered_extension_stream_limit(limit: usize) { + let mut pair = Pair::default(); + let mut server = pair.server(); + let (rejected_tx, rejected_rx) = oneshot::channel(); + let (claimed_tx, claimed_rx) = oneshot::channel(); + let (finished_tx, finished_rx) = oneshot::channel(); + + let server_fut = async { + let mut builder = server::builder(); + builder + .enable_webtransport(true) + .max_buffered_extension_streams(limit); + let mut incoming = builder.build(server.next().await).await.unwrap(); + // Drive stream classification until the peer observes rejection. No session exists. + tokio::select! { + result = rejected_rx => result.unwrap(), + _ = future::poll_fn(|cx| { + incoming.inner.poll_accept_recv(cx).unwrap(); + assert!(incoming.inner.accepted_streams_mut().wt_uni_streams.len() <= limit); + std::task::Poll::<()>::Pending + }) => unreachable!(), + } + let streams = &mut incoming.inner.accepted_streams_mut().wt_uni_streams; + assert_eq!(streams.len(), limit); + // Claiming streams releases the shared budget. + while streams.pop().is_some() {} + claimed_tx.send(()).unwrap(); + if limit > 0 { + future::poll_fn(|cx| { + incoming.inner.poll_accept_recv(cx).unwrap(); + let streams = &incoming.inner.accepted_streams_mut().wt_uni_streams; + if streams.len() == 1 { + // A different session ID uses the same buffer budget. + assert_eq!( + streams[0].0, + crate::webtransport::SessionId::from_varint(VarInt(68)) + ); + std::task::Poll::Ready(()) + } else { + std::task::Poll::Pending + } + }) + .await; + } + finished_tx.send(()).unwrap(); + }; + let client_fut = async { + let conn = pair.client_inner().await; + // Keep a valid HTTP/3 control stream open throughout the test. + let mut control = conn.open_uni().await.unwrap(); + control.write_all(b"\x00\x04\x00").await.unwrap(); + for i in 0..limit { + let mut stream = conn.open_uni().await.unwrap(); + // Valid two-byte stream type and session ID, followed by FIN. + stream + .write_all(&[0x40, 0x54, 0x40, 64 + (i % 16) as u8 * 4]) + .await + .unwrap(); + stream.finish().unwrap(); + } + // Leave excess streams open so STOP_SENDING's exact error is observable. + for _ in 0..3 { + let mut excess = conn.open_uni().await.unwrap(); + excess.write_all(b"\x40\x54\x40\x60").await.unwrap(); + // Keep data arriving while the header parser resolves both integers. + // Observe the peer's STOP_SENDING through the next write. + loop { + match excess.write_all(&[0]).await { + Ok(()) => tokio::task::yield_now().await, + Err(quinn::WriteError::Stopped(code)) => { + assert_eq!( + code.into_inner(), + Code::H3_WEBTRANSPORT_BUFFERED_STREAM_REJECTED.value() + ); + break; + } + Err(err) => panic!("unexpected write error: {err}"), + } + } + } + rejected_tx.send(()).unwrap(); + claimed_rx.await.unwrap(); + if limit > 0 { + let mut stream = conn.open_uni().await.unwrap(); + stream.write_all(b"\x40\x54\x40\x44").await.unwrap(); + stream.finish().unwrap(); + } + finished_rx.await.unwrap(); + }; + tokio::time::timeout(Duration::from_secs(10), async { + tokio::join!(server_fut, client_fut); + }) + .await + .expect("extension stream limit test timed out"); +}