feat: add max_buffered_extension_streams() config option

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.

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<Config> 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(),

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

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: <https://datatracker.ietf.org/doc/html/draft-ietf-webtrans-http3-02#section-4.5>
+    (0x3994bd84, H3_WEBTRANSPORT_BUFFERED_STREAM_REJECTED);
+
     /// Datagram or capsule parse error
     /// See: <https://www.rfc-editor.org/rfc/rfc9297#section-5.2>
     (0x33, H3_DATAGRAM_ERROR);

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;

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");
+}