Accept negotiated server-initiated streams in the H3 client

h3/src/client/builder.rs

@@ -160,6 +160,7 @@ impl Builder {
         Ok((
             Connection {
                 inner,
+                server_streams: None,
                 sent_closing: None,
                 recv_closing: None,
             },

h3/src/client/connection.rs

@@ -2,6 +2,7 @@
 
 use std::{
     marker::PhantomData,
+    num::NonZeroUsize,
     sync::{atomic::AtomicUsize, Arc},
     task::{Context, Poll},
 };
@@ -25,9 +26,10 @@ use crate::{
     quic::{self, StreamId},
     shared_state::{ConnectionState, SharedState},
     stream::{self, BufRecvStream},
+    Settings,
 };
 
-use super::stream::RequestStream;
+use super::{incoming::Dispatch, stream::RequestStream, IncomingStreams};
 
 /// HTTP/3 request sender
 ///
@@ -294,6 +296,7 @@ where
 /// #    C: quic::Connection<B> + Send + 'static,
 /// #    C::SendStream: Send + 'static,
 /// #    C::RecvStream: Send + 'static,
+/// #    C::BidiStream: Send + 'static,
 /// #    B: Buf + Send + 'static,
 /// # {
 /// // Run the driver on a different task
@@ -319,6 +322,7 @@ where
 /// #    C: quic::Connection<B> + Send + 'static,
 /// #    C::SendStream: Send + 'static,
 /// #    C::RecvStream: Send + 'static,
+/// #    C::BidiStream: Send + 'static,
 /// #    B: Buf + Send + 'static,
 /// # {
 /// // Prepare a channel to stop the driver thread
@@ -358,6 +362,7 @@ where
 {
     /// TODO: breaking encapsulation for RFC9298.
     pub inner: ConnectionInner<C, B>,
+    pub(super) server_streams: Option<Dispatch<C::BidiStream, B>>,
     // Has a GOAWAY frame been sent? If so, this PushId is the last we are willing to accept.
     pub(super) sent_closing: Option<PushId>,
     // Has a GOAWAY frame been received? If so, this is StreamId the last the remote will accept.
@@ -379,6 +384,31 @@ where
     C: quic::Connection<B>,
     B: Buf,
 {
+    /// Register bounded delivery of server-initiated streams for an HTTP/3 extension.
+    ///
+    /// `negotiated` receives the local advertised settings and the actual peer settings.
+    /// It must return true only when both endpoints have negotiated an extension that
+    /// permits these streams. It is never called before peer SETTINGS arrive. Returning
+    /// false, or receiving streams without registration, closes the connection with
+    /// `H3_STREAM_CREATION_ERROR` as required by RFC 9114 section 6.1.
+    ///
+    /// The driver never waits for queue space. Excess streams are reset and stopped with
+    /// `rejection_code`, as are queued streams abandoned by receiver drop, registration
+    /// replacement, driver drop, or driver termination. Dropping the receiver keeps the
+    /// negotiation gate installed, but rejects all later negotiated streams individually.
+    /// Replacing registration ends the old receiver. Delivered streams belong to the
+    /// extension, which must validate their framing and association.
+    pub fn server_initiated_streams(
+        &mut self,
+        capacity: NonZeroUsize,
+        rejection_code: Code,
+        negotiated: fn(&Settings, &Settings) -> bool,
+    ) -> IncomingStreams<C::BidiStream, B> {
+        let (dispatch, incoming) = Dispatch::new(capacity, rejection_code, negotiated);
+        self.server_streams = Some(dispatch);
+        incoming
+    }
+
     /// Initiate a graceful shutdown, accepting `max_push` potentially in-flight server pushes
     #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
     pub async fn shutdown(&mut self, _max_push: usize) -> Result<(), ConnectionError> {
@@ -395,7 +425,19 @@ where
     /// Maintain the connection state until it is closed
     #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
     pub fn poll_close(&mut self, cx: &mut Context<'_>) -> Poll<ConnectionError> {
-        while let Poll::Ready(result) = self.inner.poll_control(cx) {
+        let result = self.poll_close_inner(cx);
+        if result.is_ready() {
+            self.server_streams = None;
+        }
+        result
+    }
+
+    fn poll_close_inner(&mut self, cx: &mut Context<'_>) -> Poll<ConnectionError> {
+        const POLL_BUDGET: usize = 32;
+        for index in 0..POLL_BUDGET {
+            let Poll::Ready(result) = self.inner.poll_control(cx) else {
+                break;
+            };
             match result {
                 //= https://www.rfc-editor.org/rfc/rfc9114#section-7.2.4.2
                 //= type=TODO
@@ -464,14 +506,34 @@ where
                     return Poll::Ready(connection_error);
                 }
             }
+            if index + 1 == POLL_BUDGET {
+                cx.waker().wake_by_ref();
+            }
         }
 
         //= https://www.rfc-editor.org/rfc/rfc9114#section-6.1
         //# Clients MUST treat
         //# receipt of a server-initiated bidirectional stream as a connection
         //# error of type H3_STREAM_CREATION_ERROR unless such an extension has
         //# been negotiated.
-        if self.inner.poll_accept_bi(cx).is_ready() {
+        for _ in 0..POLL_BUDGET {
+            let mut stream = match self.inner.poll_accept_bi(cx) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Err(error)) => return Poll::Ready(error),
+                Poll::Ready(Ok(stream)) => stream,
+            };
+            if let Some(dispatch) = &self.server_streams {
+                if self
+                    .peer_settings()
+                    .is_some_and(|peer| (dispatch.negotiated)(&self.inner.config.settings, peer))
+                {
+                    dispatch.send(stream);
+                    continue;
+                }
+            }
+            use crate::quic::{RecvStream, SendStream};
+            stream.reset(Code::H3_STREAM_CREATION_ERROR.value());
+            stream.stop_sending(Code::H3_STREAM_CREATION_ERROR.value());
             return Poll::Ready(
                 self.inner
                     .handle_connection_error(InternalConnectionError::new(
@@ -481,6 +543,9 @@ where
             );
         }
 
+        // Yield even when the peer always has another stream ready. Control processing
+        // runs first on the next poll; no queue capacity wait can stall the driver.
+        cx.waker().wake_by_ref();
         Poll::Pending
     }
 }

h3/src/client/incoming.rs

@@ -0,0 +1,131 @@
+use std::{
+    collections::VecDeque,
+    marker::PhantomData,
+    num::NonZeroUsize,
+    sync::{Arc, Mutex},
+    task::{Context, Poll},
+};
+
+use bytes::Buf;
+use futures_util::{future::poll_fn, task::AtomicWaker};
+
+use crate::{error::Code, quic, Settings};
+
+/// Bounded receiver of raw server-initiated bidirectional extension streams.
+///
+/// The connection driver must keep running. The extension must validate each stream's
+/// prefix and association before using it. Once returned, a stream belongs to the
+/// caller, including responsibility for resetting and stopping abandoned streams.
+/// Dropping this receiver aborts queued streams and makes the driver reject new ones.
+pub struct IncomingStreams<S: quic::SendStream<B> + quic::RecvStream, B: Buf> {
+    shared: Arc<Shared<S, B>>,
+}
+
+struct Shared<S: quic::SendStream<B> + quic::RecvStream, B: Buf> {
+    queue: Mutex<Queue<S>>,
+    waker: AtomicWaker,
+    capacity: NonZeroUsize,
+    rejection_code: Code,
+    _buf: PhantomData<fn(B)>,
+}
+
+struct Queue<S> {
+    streams: VecDeque<S>,
+    open: bool,
+}
+
+impl<S: quic::SendStream<B> + quic::RecvStream, B: Buf> Shared<S, B> {
+    fn reject(&self, mut stream: S) {
+        stream.reset(self.rejection_code.value());
+        stream.stop_sending(self.rejection_code.value());
+    }
+
+    fn close(&self) {
+        let streams = {
+            let mut queue = self.queue.lock().unwrap();
+            queue.open = false;
+            std::mem::take(&mut queue.streams)
+        };
+        for stream in streams {
+            self.reject(stream);
+        }
+        self.waker.wake();
+    }
+}
+
+impl<S: quic::SendStream<B> + quic::RecvStream, B: Buf> IncomingStreams<S, B> {
+    /// Receive the next stream, or `None` when the driver ends or registration is replaced.
+    ///
+    /// Cancelling this future leaves queued streams available to the next call.
+    pub async fn recv(&mut self) -> Option<S> {
+        poll_fn(|cx| self.poll_recv(cx)).await
+    }
+
+    /// Poll for the next stream. Only the most recent caller's waker is retained.
+    pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<S>> {
+        self.shared.waker.register(cx.waker());
+        let mut queue = self.shared.queue.lock().unwrap();
+        if let Some(stream) = queue.streams.pop_front() {
+            Poll::Ready(Some(stream))
+        } else if queue.open {
+            Poll::Pending
+        } else {
+            Poll::Ready(None)
+        }
+    }
+}
+
+impl<S: quic::SendStream<B> + quic::RecvStream, B: Buf> Drop for IncomingStreams<S, B> {
+    fn drop(&mut self) {
+        self.shared.close();
+    }
+}
+
+pub(super) struct Dispatch<S: quic::SendStream<B> + quic::RecvStream, B: Buf> {
+    shared: Arc<Shared<S, B>>,
+    pub(super) negotiated: fn(&Settings, &Settings) -> bool,
+}
+
+impl<S: quic::SendStream<B> + quic::RecvStream, B: Buf> Dispatch<S, B> {
+    pub(super) fn new(
+        capacity: NonZeroUsize,
+        rejection_code: Code,
+        negotiated: fn(&Settings, &Settings) -> bool,
+    ) -> (Self, IncomingStreams<S, B>) {
+        let shared = Arc::new(Shared {
+            queue: Mutex::new(Queue {
+                streams: VecDeque::new(),
+                open: true,
+            }),
+            capacity,
+            rejection_code,
+            waker: AtomicWaker::new(),
+            _buf: PhantomData,
+        });
+        (
+            Self {
+                shared: shared.clone(),
+                negotiated,
+            },
+            IncomingStreams { shared },
+        )
+    }
+
+    pub(super) fn send(&self, stream: S) {
+        let mut queue = self.shared.queue.lock().unwrap();
+        if queue.open && queue.streams.len() < self.shared.capacity.get() {
+            queue.streams.push_back(stream);
+            drop(queue);
+            self.shared.waker.wake();
+        } else {
+            drop(queue);
+            self.shared.reject(stream);
+        }
+    }
+}
+
+impl<S: quic::SendStream<B> + quic::RecvStream, B: Buf> Drop for Dispatch<S, B> {
+    fn drop(&mut self) {
+        self.shared.close();
+    }
+}

h3/src/client/mod.rs

@@ -1,6 +1,7 @@
 //! HTTP/3 client
 
 mod connection;
+mod incoming;
 mod stream;
 
 mod builder;
@@ -9,4 +10,5 @@ pub use builder::builder;
 pub use builder::new;
 pub use builder::Builder;
 pub use connection::{Connection, SendRequest};
+pub use incoming::IncomingStreams;
 pub use stream::RequestStream;

h3/src/lib.rs

@@ -5,6 +5,7 @@
 pub mod client;
 
 mod config;
+pub use config::Settings;
 //pub mod error;
 pub mod ext;
 pub mod quic;

h3/src/tests/mod.rs

@@ -11,6 +11,7 @@ mod h3_quinn;
 
 mod connection;
 mod request;
+mod server_initiated;
 
 use std::{
     convert::TryInto,