refactor(quic)!: one Limits type for both QUIC runtimes

doc/bin/relay/config.md

@@ -56,14 +56,14 @@ Transport tuning, applied to accepted and dialed connections alike.
 ```toml
 [quic]
 congestion_control = "delay"         # "delay" (BBR, the default) or "loss" (CUBIC).
-max_streams = 10000                  # Concurrent streams per connection, bidi and uni. Default.
+max_streams = 10000                  # Concurrent streams per connection, bidi and uni; 0 is refused. Default.
 idle_timeout = "10s"                 # Drop a connection after this long with nothing on it. Default.
 keep_alive = "3s"                    # Ping interval; "0s" disables it. iroh still pings at least every 5s. Default.
 gso = true                           # UDP segmentation offload. iroh cannot turn it off.
 mtu_discovery = false                # Path MTU discovery. Default.
 receive_window = 67108864            # Whole-connection window, in bytes. Default (64 MiB).
-stream_receive_window = 8388608      # Per-stream window. Omit for the backend default.
-send_window = 33554432               # Unacknowledged outgoing data. Omit for the backend default.
+stream_receive_window = 4194304      # Per-stream window, in bytes. Default (4 MiB).
+send_window = 16777216               # Unacknowledged outgoing data, in bytes. Default (16 MiB).
 qlog = "/var/log/moq/qlog"           # Existing directory. Needs the `qlog` build feature.
 ```
 

go/wrapper/client.go

@@ -227,7 +227,7 @@ func WithResolutionDelay(delay time.Duration) ClientOption {
 }
 
 // WithQUICMaxStreams caps the concurrent QUIC streams the peer may open toward
-// this connection. MoQ opens a stream per group, and for a
+// this connection; 0 is refused. MoQ opens a stream per group, and for a
 // subscriber those arrive from the relay, so a client subscribing to many tracks
 // wants this raised. A publisher's own streams are bounded by the relay's
 // advertised limit, not this one. Ignored by the WebSocket fallback.

go/wrapper/server.go

@@ -121,7 +121,7 @@ func WithServerVersions(versions ...string) ServerOption {
 }
 
 // WithServerQUICMaxStreams caps the concurrent QUIC streams each peer may open
-// toward this server (default 1024).
+// toward this server (default 10,000; 0 is refused).
 func WithServerQUICMaxStreams(maxStreams uint64) ServerOption {
 	return func(c *serverConfig) { c.quicMaxStreams = &maxStreams }
 }

quest/m2/rate-quic.md

@@ -59,9 +59,9 @@ backpressure the peer cannot ignore, rather than metering and closing.
   rides the CONNECT URL or arrives in band), so it is set on a live
   connection and reset on revalidation or a union change. Extend the
   runtime limits seam the relay widens cluster peers through
-  (`moq_tokio::server::Request::set_quic_limits` and
-  `moq_uring::quic::web::Session::set_limits`, each taking its backend's
-  `quic::Limits`) with the two rates, and reach it from the live session
+  (`set_limits` on `moq_tokio::server::Request::quic` and on
+  `moq_uring::quic::web::Session`, both taking `moq_quic::Limits`) with the
+  two rates, and reach it from the live session
   so revalidation can change it after SETUP.
 - **Unions take the highest cap.** When in-band auth unions several grants
   on one session, each direction's cap is the highest among them, and an

rs/moq-ffi/src/session.rs

@@ -99,7 +99,7 @@ pub struct MoqClientTls {
 /// one; `gso`, `mtu_discovery`, and `congestion_control` leave the choice to the backend.
 #[derive(Clone, Debug, Default, PartialEq, Eq, uniffi::Record)]
 pub struct MoqQuicConfig {
-	/// Cap on the concurrent QUIC streams the peer may open toward this endpoint.
+	/// Cap on the concurrent QUIC streams the peer may open toward this endpoint; 0 is refused.
 	///
 	/// MoQ opens a stream per group, and for a subscriber those arrive from the peer, so an
 	/// endpoint subscribing to many tracks wants this raised.

rs/moq-quic/README.md

@@ -73,6 +73,7 @@ Changes on top of the upstream commit, besides the renames:
 - **Fuzzing** runs through the hidden `fuzz` module behind a `fuzz` feature, with one-line shims in [`fuzz/`](fuzz/README.md): upstream's `packet`, `params`, `streamid`, and `streams` targets, plus `frame` and a connection op sequence. Upstream's `cfg(fuzzing)` `fuzzing` module and `arbitrary` feature are gone, so its fuzz patches port by hand.
 - **Draining endpoint event**: `EndpointEvent::is_draining` fires once per connection when it stops being active, before the drained event, so the driver's `Endpoint::wait_all_draining` can end a shutdown without sitting out every closing period.
 - **Late stream limits**: peer transport parameters that raise a stream limit after an open was refused emit `StreamEvent::Available`, as `MAX_STREAMS` does. A server accepted before the rest of a multi-packet ClientHello (a post-quantum key share is enough) otherwise leaves a 0.5-RTT `open_uni` parked until the idle timeout. Not offered upstream yet.
+- **Connection limits** (`Limits`, `Connection::set_limits`, and the drivers' `set_limits`): one validated call changes a live connection's stream limit and windows, refusing values a peer would reject and changing nothing then. moq-tokio and moq-uring re-export the type, so both runtimes check limits the same way. Not offered upstream: it only bundles quinn's existing setters.
 
 Recheck the BBR3 changes when cherry-picking quinn#2481 updates, or if it merges.
 

rs/moq-quic/src/connection/mod.rs

@@ -15,9 +15,9 @@ use thiserror::Error;
 use tracing::{debug, error, trace, trace_span, warn};
 
 use crate::{
-    Dir, Duration, EndpointConfig, Frame, Instant, MAX_CID_SIZE, MAX_STREAM_COUNT,
-    MIN_INITIAL_SIZE, Side, StreamId, TIMER_GRANULARITY, TokenStore, Transmit, TransportError,
-    TransportErrorCode, VarInt,
+    Dir, Duration, EndpointConfig, Frame, Instant, Limits, LimitsError, MAX_CID_SIZE,
+    MAX_STREAM_COUNT, MIN_INITIAL_SIZE, Side, StreamId, TIMER_GRANULARITY, TokenStore, Transmit,
+    TransportError, TransportErrorCode, VarInt,
     cid_queue::CidQueue,
     coding::BufMutExt,
     config::{ServerConfig, TransportConfig},
@@ -1570,6 +1570,25 @@ impl Connection {
         }
     }
 
+    /// Change every limit `limits` sets, leaving the others as they are
+    ///
+    /// Errors, changing nothing, on limits [`Limits::validate`] refuses.
+    pub fn set_limits(&mut self, limits: Limits) -> Result<(), LimitsError> {
+        limits.validate()?;
+        if let Some(window) = limits.receive_window {
+            self.set_receive_window(VarInt::from_u64(window.get()).expect("validated"));
+        }
+        if let Some(window) = limits.send_window {
+            self.set_send_window(window.get());
+        }
+        if let Some(streams) = limits.max_streams {
+            let streams = VarInt::from_u64(streams.get()).expect("validated");
+            self.set_max_concurrent_streams(Dir::Bi, streams);
+            self.set_max_concurrent_streams(Dir::Uni, streams);
+        }
+        Ok(())
+    }
+
     fn on_ack_received(
         &mut self,
         now: Instant,

rs/moq-quic/src/lib.rs

@@ -65,12 +65,14 @@ pub use connection::qlog::QlogStream;
 pub use rustls;
 
 mod config;
+mod limits;
 #[cfg(feature = "qlog")]
 pub use config::QlogConfig;
 pub use config::{
     AckFrequencyConfig, ClientConfig, ConfigError, EndpointConfig, IdleTimeout, MtuDiscoveryConfig,
     ServerConfig, StdSystemTime, TimeSource, TransportConfig, ValidationTokenConfig,
 };
+pub use limits::{Limits, LimitsError};
 
 pub mod crypto;
 

rs/moq-quic/src/limits.rs

@@ -0,0 +1,88 @@
+use std::num::NonZeroU64;
+
+use thiserror::Error;
+
+use crate::{MAX_STREAM_COUNT, VarInt};
+
+/// Flow-control limits an established connection can change, such as a relay widening them
+/// once it knows the peer is another relay
+///
+/// Each limit is `None` to leave it as it is. Each is non-zero because zero streams or a zero
+/// window credits nothing and wedges the connection. Build one from [`Limits::default`] and
+/// set the fields to change.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+#[non_exhaustive]
+pub struct Limits {
+    /// Max concurrent streams the peer may open, per direction
+    pub max_streams: Option<NonZeroU64>,
+    /// Connection-wide receive window, in bytes
+    pub receive_window: Option<NonZeroU64>,
+    /// Cap on unacknowledged outgoing data, in bytes
+    pub send_window: Option<NonZeroU64>,
+}
+
+impl Limits {
+    /// Refuse the limits a peer would reject: a stream count past 2^60, or a receive window
+    /// past a QUIC varint
+    pub fn validate(&self) -> Result<(), LimitsError> {
+        if self
+            .max_streams
+            .is_some_and(|streams| streams.get() > MAX_STREAM_COUNT)
+        {
+            return Err(LimitsError::MaxStreams);
+        }
+        if self
+            .receive_window
+            .is_some_and(|window| VarInt::from_u64(window.get()).is_err())
+        {
+            return Err(LimitsError::ReceiveWindow);
+        }
+        Ok(())
+    }
+}
+
+/// Why [`Limits`] cannot be applied
+#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum LimitsError {
+    /// The stream limit is past the 2^60 a peer accepts
+    #[error("max_streams must be at most 2^60")]
+    MaxStreams,
+    /// The receive window does not fit a QUIC varint
+    #[error("receive_window must fit a QUIC varint")]
+    ReceiveWindow,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn limits(max_streams: u64, receive_window: u64, send_window: u64) -> Limits {
+        Limits {
+            max_streams: NonZeroU64::new(max_streams),
+            receive_window: NonZeroU64::new(receive_window),
+            send_window: NonZeroU64::new(send_window),
+        }
+    }
+
+    #[test]
+    fn validate_accepts_the_wire_maximums() {
+        assert_eq!(Limits::default().validate(), Ok(()));
+        assert_eq!(
+            limits(MAX_STREAM_COUNT, VarInt::MAX.into_inner(), u64::MAX).validate(),
+            Ok(())
+        );
+    }
+
+    #[test]
+    fn validate_refuses_what_a_peer_would_reject() {
+        assert_eq!(
+            limits(MAX_STREAM_COUNT + 1, 1, 1).validate(),
+            Err(LimitsError::MaxStreams)
+        );
+        assert_eq!(
+            limits(1, VarInt::MAX.into_inner() + 1, 1).validate(),
+            Err(LimitsError::ReceiveWindow)
+        );
+    }
+}

rs/moq-quic/src/tests/characterize/limits.rs

@@ -1,5 +1,7 @@
 //! Stream and flow control limits changed while a connection is up
 
+use std::num::NonZeroU64;
+
 use super::*;
 
 /// How many more uni streams the client can open right now, opening them all
@@ -239,3 +241,54 @@ fn zero_rtt_bidi_window_raise() {
 fn zero_rtt_uni_window_raise() {
     zero_rtt_window_raise(Dir::Uni);
 }
+
+/// `set_limits` raises every limit it sets in one call
+#[test]
+fn set_limits_raises_what_it_sets() {
+    let _guard = subscribe();
+    let (mut pair, client_ch, server_ch) = connected_with(
+        |t| {
+            t.max_concurrent_uni_streams(VarInt(1));
+            t.receive_window(VarInt::from_u32(WINDOW));
+        },
+        |_| {},
+    );
+
+    let limits = Limits {
+        max_streams: NonZeroU64::new(3),
+        receive_window: NonZeroU64::new(2 * WINDOW as u64),
+        ..Default::default()
+    };
+    pair.server_conn_mut(server_ch).set_limits(limits).unwrap();
+    pair.drive();
+
+    // One stream to fill the wider window, then the two the raised limit leaves
+    assert_eq!(
+        writable(&mut pair, client_ch, 4 * WINDOW as usize),
+        2 * WINDOW as usize
+    );
+    assert_eq!(openable(&mut pair, client_ch).len(), 2);
+}
+
+/// Limits a peer would reject change nothing, rather than applying the ones that fit
+#[test]
+fn set_limits_refuses_without_changing_anything() {
+    let _guard = subscribe();
+    let (mut pair, _, server_ch) =
+        connected_with(|t| _ = t.max_concurrent_uni_streams(VarInt(1)), |_| {});
+
+    let limits = Limits {
+        max_streams: NonZeroU64::new(3),
+        receive_window: NonZeroU64::new(1 << 62),
+        ..Default::default()
+    };
+    assert_eq!(
+        pair.server_conn_mut(server_ch).set_limits(limits),
+        Err(LimitsError::ReceiveWindow)
+    );
+    assert_eq!(
+        pair.server_conn_mut(server_ch)
+            .max_concurrent_streams(Dir::Uni),
+        1
+    );
+}

rs/moq-relay/Cargo.toml

@@ -30,7 +30,7 @@ iroh = ["moq-tokio/iroh"]
 # Discover cluster peers on the LAN with mDNS (`--cluster-lan`).
 cluster-lan = ["moq-tokio/mdns"]
 jemalloc = ["moq-tokio/jemalloc"]
-quic = ["_quic", "moq-tokio/quic"]
+quic = ["_quic", "moq-tokio/quic", "dep:moq-quic"]
 # The tokio QUIC listener, and the thread-per-core worker group behind
 # `runtime.workers`. Not one to enable on its own: it names no backend, so
 # moq-tokio would have nothing to bind a socket with.
@@ -48,7 +48,7 @@ uds = ["moq-tokio/uds"]
 # This is off by default: the relay is useful without io_uring.
 io-uring = ["_uring", "moq-uring/quic"]
 # The listener machinery shared by the feature above.
-_uring = ["dep:moq-sock", "dep:moq-uring"]
+_uring = ["dep:moq-sock", "dep:moq-uring", "dep:moq-quic"]
 
 [dependencies]
 anyhow = { workspace = true, features = ["backtrace"] }
@@ -66,6 +66,8 @@ humantime = { workspace = true }
 hyper-util = { workspace = true }
 moq-auth = { workspace = true, features = ["client"] }
 moq-net = { workspace = true }
+# The flow-control limits both QUIC runtimes take, for widening a peer's connection.
+moq-quic = { workspace = true, optional = true }
 moq-stats = { workspace = true }
 moq-tokio = { workspace = true, default-features = false, features = ["aws-lc-rs", "watch", "tcp"] }
 qmux = { workspace = true, features = ["ws"], optional = true }

rs/moq-relay/src/cluster.rs

@@ -1089,7 +1089,7 @@ struct Work {
 pub struct Cluster {
 	session_limits: moq_net::session::Limits,
 	peer_limits: moq_net::session::Limits,
-	peer_quic: moq_tokio::quic::Limits,
+	peer_quic: crate::limits::Quic,
 	config: Config,
 	/// Dial template every peer dial is built from; see [`Self::peer_client`].
 	connect: Option<moq_tokio::connect::Config>,
@@ -1171,8 +1171,9 @@ impl Cluster {
 	}
 
 	/// The QUIC limits an admitted session widens to, if any: only a `peer` grant does.
-	pub(crate) fn quic_limits(&self, lease: &auth::Lease) -> Option<moq_tokio::quic::Limits> {
-		lease.token().peer.then_some(self.peer_quic)
+	#[cfg(any(feature = "quic", all(target_os = "linux", feature = "_uring")))]
+	pub(crate) fn quic_limits(&self, lease: &auth::Lease) -> Option<moq_quic::Limits> {
+		lease.token().peer.then(|| self.peer_quic.limits())
 	}
 
 	/// The origin ID used by this relay on the wire.
@@ -1216,6 +1217,7 @@ impl Cluster {
 		// matching the `--cluster-connect-api` validation, instead of silently
 		// keeping the first entry when dials spawn.
 		parse_peer_list(config.connect.clone(), None).context("invalid --cluster-connect peer list")?;
+		peer.quic.validate()?;
 		let critical = crate::critical::Critical::default();
 		let mut origin_config = origin::Config::new(id);
 		if let Some(cache) = cache {
@@ -1234,7 +1236,7 @@ impl Cluster {
 		Ok(Cluster {
 			session_limits: session.resolve(),
 			peer_limits: peer.resolve(),
-			peer_quic: peer.quic.resolve()?,
+			peer_quic: peer.quic,
 			config,
 			connect: None,
 			quic: None,
@@ -2086,9 +2088,7 @@ impl Cluster {
 			connect.tls.fingerprint = vec![fingerprint.to_string()];
 		}
 		let mut quic = self.quic.clone().unwrap_or_default();
-		quic.max_streams = Some(self.peer_quic.max_streams.get());
-		quic.receive_window = Some(self.peer_quic.receive_window.get());
-		quic.send_window = Some(self.peer_quic.send_window.get());
+		self.peer_quic.dial(&mut quic);
 		let client = connect.init(quic)?;
 		#[cfg(feature = "iroh")]
 		let client = match &self.iroh {

rs/moq-relay/src/connection.rs

@@ -109,10 +109,12 @@ impl Connection {
 		} = admitted;
 
 		// A peer's QUIC limits widen before SETUP OK, so its first requests see them.
+		// Other transports keep their own limits.
+		#[cfg(feature = "quic")]
 		if let Some(limits) = lifecycle.cluster.quic_limits(&lease)
-			&& let Err(err) = request.set_quic_limits(limits)
+			&& let Some(quic) = request.quic()
 		{
-			tracing::debug!(%err, "cluster peer keeps its transport's limits");
+			quic.set_limits(limits)?;
 		}
 
 		// Accept the connection.