From 3a7caef2c969afeea7324c9cad538d3bbe42f60a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 08:24:40 -0700 Subject: [PATCH 1/2] refactor(quic)!: one Limits type for both QUIC runtimes moq-quic owns `Limits` (validated, `#[non_exhaustive]`, `None` leaves a limit unchanged) and `Connection::set_limits`. moq-tokio and moq-uring re-export it as `quic::Limits`, and both expose `quic::Connection::set_limits(Limits) -> Result<(), quic::Error>`. `Request::quic()` replaces `set_quic_limits` and `Error::QuicLimits`. `[quic] max_streams = 0` is refused, and the connection receive window default resolves in `quic::Config::resolve`, so io_uring workers run the documented 64 MiB instead of 16 MiB. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + cpp/obs/src/moq-settings.cpp | 2 +- doc/bin/relay/config.md | 2 +- go/wrapper/client.go | 9 +- go/wrapper/server.go | 2 +- quest/m2/rate-quic.md | 6 +- rs/moq-ffi/src/session.rs | 3 +- rs/moq-quic/README.md | 1 + rs/moq-quic/src/connection/mod.rs | 25 +++- rs/moq-quic/src/lib.rs | 2 + rs/moq-quic/src/limits.rs | 88 ++++++++++++++ rs/moq-quic/src/tests/characterize/limits.rs | 53 +++++++++ rs/moq-relay/Cargo.toml | 6 +- rs/moq-relay/src/cluster.rs | 14 +-- rs/moq-relay/src/connection.rs | 6 +- rs/moq-relay/src/limits.rs | 29 +++-- rs/moq-relay/src/uring.rs | 22 ++-- rs/moq-relay/tests/integration/peer_limits.rs | 14 ++- rs/moq-tokio/src/error.rs | 9 +- rs/moq-tokio/src/iroh.rs | 4 +- rs/moq-tokio/src/quic.rs | 107 +++++++----------- rs/moq-tokio/src/quic/backend.rs | 28 ++++- rs/moq-tokio/src/quic/driver/connection.rs | 9 ++ rs/moq-tokio/src/server.rs | 35 ++---- rs/moq-uring/src/error.rs | 8 -- rs/moq-uring/src/quic/driver/connection.rs | 22 +--- rs/moq-uring/src/quic/mod.rs | 21 +--- rs/moq-uring/src/quic/web.rs | 2 +- 28 files changed, 333 insertions(+), 197 deletions(-) create mode 100644 rs/moq-quic/src/limits.rs diff --git a/Cargo.lock b/Cargo.lock index 0bff579d024..ff05c33f655 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4893,6 +4893,7 @@ dependencies = [ "libc", "moq-auth 0.2.1", "moq-net", + "moq-quic", "moq-shaper", "moq-sock", "moq-stats", diff --git a/cpp/obs/src/moq-settings.cpp b/cpp/obs/src/moq-settings.cpp index 4713bc61c1a..30d49aa9ff5 100644 --- a/cpp/obs/src/moq-settings.cpp +++ b/cpp/obs/src/moq-settings.cpp @@ -33,7 +33,7 @@ constexpr const char *WEBSOCKET_DELAY = "websocket_delay_ms"; constexpr long long BACKOFF_INITIAL_DEFAULT_MS = 1000; constexpr long long BACKOFF_MAX_DEFAULT_MS = 5000; constexpr long long BACKOFF_TIMEOUT_DEFAULT_MS = 10000; -constexpr long long QUIC_MAX_STREAMS_DEFAULT = 1024; +constexpr long long QUIC_MAX_STREAMS_DEFAULT = 10000; constexpr bool WEBSOCKET_ENABLED_DEFAULT = true; constexpr long long WEBSOCKET_DELAY_DEFAULT_MS = 200; diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index f0115c6632b..4fdb821c24b 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -56,7 +56,7 @@ 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. diff --git a/go/wrapper/client.go b/go/wrapper/client.go index caf04525848..e555e206d69 100644 --- a/go/wrapper/client.go +++ b/go/wrapper/client.go @@ -173,10 +173,11 @@ func WithVersions(versions ...string) ClientOption { } // WithQUICMaxStreams caps the concurrent QUIC streams the peer may open toward -// this connection (default 1024). 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. +// this connection (default 10,000; 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. func WithQUICMaxStreams(maxStreams uint64) ClientOption { return func(c *clientConfig) { c.quicMaxStreams = &maxStreams } } diff --git a/go/wrapper/server.go b/go/wrapper/server.go index a3fc9e4854e..20a6f84fc15 100644 --- a/go/wrapper/server.go +++ b/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 } } diff --git a/quest/m2/rate-quic.md b/quest/m2/rate-quic.md index 9e5c57c74d9..a77910c7897 100644 --- a/quest/m2/rate-quic.md +++ b/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 diff --git a/rs/moq-ffi/src/session.rs b/rs/moq-ffi/src/session.rs index 1d792706e2e..a0cf2c137de 100644 --- a/rs/moq-ffi/src/session.rs +++ b/rs/moq-ffi/src/session.rs @@ -81,7 +81,8 @@ pub struct MoqClientTls { /// QUIC transport tuning, mirroring moq-tokio's QUIC config. #[derive(Clone, Debug, Default, PartialEq, Eq, uniffi::Record)] pub struct MoqQuicConfig { - /// Cap on the concurrent QUIC streams the peer may open toward this endpoint. Null uses 1024. + /// Cap on the concurrent QUIC streams the peer may open toward this endpoint. Null uses 10,000, + /// and 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. Ignored by the WebSocket fallback. diff --git a/rs/moq-quic/README.md b/rs/moq-quic/README.md index 8d12d1e703a..5ea0b1902b3 100644 --- a/rs/moq-quic/README.md +++ b/rs/moq-quic/README.md @@ -72,6 +72,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. diff --git a/rs/moq-quic/src/connection/mod.rs b/rs/moq-quic/src/connection/mod.rs index b2f5fdb6049..3fb0aa70e4a 100644 --- a/rs/moq-quic/src/connection/mod.rs +++ b/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}, @@ -1558,6 +1558,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, diff --git a/rs/moq-quic/src/lib.rs b/rs/moq-quic/src/lib.rs index 4c30d634a1e..1bbf7520997 100644 --- a/rs/moq-quic/src/lib.rs +++ b/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; diff --git a/rs/moq-quic/src/limits.rs b/rs/moq-quic/src/limits.rs new file mode 100644 index 00000000000..877f8a59dcd --- /dev/null +++ b/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, + /// Connection-wide receive window, in bytes + pub receive_window: Option, + /// Cap on unacknowledged outgoing data, in bytes + pub send_window: Option, +} + +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) + ); + } +} diff --git a/rs/moq-quic/src/tests/characterize/limits.rs b/rs/moq-quic/src/tests/characterize/limits.rs index 61f26367b15..6fe929ba6ae 100644 --- a/rs/moq-quic/src/tests/characterize/limits.rs +++ b/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 + ); +} diff --git a/rs/moq-relay/Cargo.toml b/rs/moq-relay/Cargo.toml index d1fd9a5c26c..98922d02bfe 100644 --- a/rs/moq-relay/Cargo.toml +++ b/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 } diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index 6014c310ac5..d056c1541ac 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/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, @@ -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 { - 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 { + 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 { diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index cced96da4b0..8e6cd726057 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -125,10 +125,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) = self.cluster.quic_limits(&lease) - && let Err(err) = self.request.set_quic_limits(limits) + && let Some(quic) = self.request.quic() { - tracing::debug!(%err, "cluster peer keeps its transport's limits"); + quic.set_limits(limits)?; } // Accept the connection. diff --git a/rs/moq-relay/src/limits.rs b/rs/moq-relay/src/limits.rs index 09aab2539ec..09988a8c96d 100644 --- a/rs/moq-relay/src/limits.rs +++ b/rs/moq-relay/src/limits.rs @@ -148,14 +148,27 @@ impl Default for Quic { impl Quic { /// Refuses limits QUIC cannot advertise, as `[quic]` does, so the accept and /// dial paths never disagree about them. - pub(crate) fn resolve(&self) -> anyhow::Result { - let limits = moq_tokio::quic::Limits { - max_streams: self.max_streams, - receive_window: self.receive_window, - send_window: self.send_window, - }; - limits.validate().context("invalid [peer.quic]")?; - Ok(limits) + pub(crate) fn validate(&self) -> anyhow::Result<()> { + let mut quic = moq_tokio::quic::Config::default(); + self.dial(&mut quic); + quic.validate().context("invalid [peer.quic]") + } + + /// Apply these limits to the `[quic]` section a peer link is dialed with. + pub(crate) fn dial(&self, quic: &mut moq_tokio::quic::Config) { + quic.max_streams = Some(self.max_streams.get()); + quic.receive_window = Some(self.receive_window.get()); + quic.send_window = Some(self.send_window.get()); + } + + /// The limits an accepted peer connection widens to, as either QUIC runtime takes them. + #[cfg(any(feature = "quic", all(target_os = "linux", feature = "_uring")))] + pub(crate) fn limits(&self) -> moq_quic::Limits { + let mut limits = moq_quic::Limits::default(); + limits.max_streams = Some(self.max_streams); + limits.receive_window = Some(self.receive_window); + limits.send_window = Some(self.send_window); + limits } } diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index a03f38f28f7..2f827eafbe3 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -423,10 +423,11 @@ fn transport(quic: &moq_tokio::quic::Resolved) -> anyhow::Result String { } } -/// A limit QUIC cannot advertise is refused at load, in `[peer.quic]` and -/// `[quic]` alike, rather than wedging the peer or the handshake later. +/// A limit QUIC cannot advertise or honor is refused at load, in `[peer.quic]` +/// and `[quic]` alike, rather than wedging the peer or the handshake later. #[tokio::test] async fn unadvertisable_quic_limits_are_refused_at_load() { let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); @@ -182,8 +182,14 @@ async fn unadvertisable_quic_limits_are_refused_at_load() { .await; assert!(err.contains("[peer.quic]") && err.contains("max_streams"), "{err}"); - let err = refused_at_load(|config| config.quic.max_streams = Some((1 << 60) + 1)).await; - assert!(err.contains("[quic]") && err.contains("max_streams"), "{err}"); + // Zero lets the peer open nothing, so it is refused too rather than wedging the connection. + for max_streams in [0, (1 << 60) + 1] { + let err = refused_at_load(|config| config.quic.max_streams = Some(max_streams)).await; + assert!( + err.contains("[quic]") && err.contains("max_streams"), + "{max_streams}: {err}" + ); + } } /// A bare QUIC server speaking the dial template's ALPNs: the dialer's stream diff --git a/rs/moq-tokio/src/error.rs b/rs/moq-tokio/src/error.rs index 497bb9a7ab8..404702e2e8f 100644 --- a/rs/moq-tokio/src/error.rs +++ b/rs/moq-tokio/src/error.rs @@ -105,14 +105,10 @@ pub enum Error { #[error("quic {0} must be non-zero and fit a QUIC varint")] WindowRange(&'static str), - /// A QUIC stream limit is past the 2^60 a peer accepts. - #[error("quic max_streams must be at most 2^60")] + /// A QUIC stream limit is zero or past the 2^60 a peer accepts. + #[error("quic max_streams must be between 1 and 2^60")] StreamsRange, - /// A session's QUIC limits were changed on a transport that is not QUIC. - #[error("{0} sessions have no QUIC limits to change")] - QuicLimits(crate::Transport), - /// Every backend we tried gave up without reporting why. #[error("failed to connect to server")] ConnectFailed, @@ -260,7 +256,6 @@ impl Error { | Self::IdleTimeoutRange | Self::WindowRange(_) | Self::StreamsRange - | Self::QuicLimits(_) | Self::QlogUnsupported | Self::Connect(_) => true, #[cfg(feature = "quic")] diff --git a/rs/moq-tokio/src/iroh.rs b/rs/moq-tokio/src/iroh.rs index 058f33ca0c9..b4256ad7722 100644 --- a/rs/moq-tokio/src/iroh.rs +++ b/rs/moq-tokio/src/iroh.rs @@ -288,9 +288,7 @@ fn transport_config(quic: &crate::quic::Resolved) -> iroh::endpoint::QuicTranspo if !quic.mtu_discovery { transport = transport.mtu_discovery_config(None); } - // iroh's connection window is unlimited, so apply ours when unset. - let window = quic.receive_window.unwrap_or(crate::quic::DEFAULT_RECEIVE_WINDOW); - let window = iroh::endpoint::VarInt::from_u64(window).unwrap_or(iroh::endpoint::VarInt::MAX); + let window = iroh::endpoint::VarInt::from_u64(quic.receive_window).unwrap_or(iroh::endpoint::VarInt::MAX); transport = transport.receive_window(window); if let Some(window) = quic.stream_receive_window { let window = iroh::endpoint::VarInt::from_u64(window).unwrap_or(iroh::endpoint::VarInt::MAX); diff --git a/rs/moq-tokio/src/quic.rs b/rs/moq-tokio/src/quic.rs index a9938bf2224..ddab67eb398 100644 --- a/rs/moq-tokio/src/quic.rs +++ b/rs/moq-tokio/src/quic.rs @@ -17,9 +17,12 @@ use crate::cli::Duration as CliDuration; #[cfg(feature = "quic")] mod backend; #[cfg(feature = "quic")] -pub use backend::Error; -#[cfg(feature = "quic")] pub(crate) use backend::{Client, Closer, Options, Server, accept}; +#[cfg(feature = "quic")] +pub use backend::{Connection, Error}; +/// The flow-control limits a live QUIC connection can change; see [`Connection::set_limits`]. +#[cfg(feature = "quic")] +pub use moq_quic::{Limits, LimitsError}; // The async QUIC layer driving `moq-quic` on tokio, forked from quinn (see its module docs). // Public only so the workspace's tests can dial a bare endpoint; not a stable API. @@ -125,33 +128,6 @@ pub(crate) const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(10); /// another (plus QUIC's probe retransmits) before the peer's timer runs out. pub(crate) const DEFAULT_KEEP_ALIVE: Duration = Duration::from_secs(3); -/// The flow-control limits a QUIC connection can be widened to, such as by a -/// relay once it knows the peer is another relay. -/// -/// Each is non-zero because zero streams or a zero window credits nothing and -/// wedges the connection. [`Limits::validate`] refuses the values QUIC cannot -/// advertise, as [`Config`] does, and every setter taking one checks it. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct Limits { - /// Max concurrent streams the peer may open, per direction. - pub max_streams: std::num::NonZeroU64, - /// Connection-wide receive window, in bytes. - pub receive_window: std::num::NonZeroU64, - /// Cap on unacknowledged outgoing data, in bytes. - pub send_window: std::num::NonZeroU64, -} - -impl Limits { - /// Refuse a stream count past 2^60 or a receive window past a QUIC varint. - pub fn validate(&self) -> crate::Result<()> { - validate_streams(Some(self.max_streams.get()))?; - match self.receive_window.get() { - window if window > MAX_VARINT => Err(crate::Error::WindowRange("receive_window")), - _ => Ok(()), - } - } -} - /// The `--quic-*` transport section, applied to dialed and accepted connections alike. #[derive(Clone, Debug, usage::Args, serde::Serialize, serde::Deserialize)] #[usage(unknown_flags = "error", args_override_self = false)] @@ -159,7 +135,8 @@ impl Limits { #[non_exhaustive] pub struct Config { /// Maximum number of concurrent QUIC streams per connection (both bidi and uni). - /// Defaults to 10,000. MoQ opens a stream per group, so busy endpoints want this high. + /// Defaults to 10,000; zero is refused. MoQ opens a stream per group, so busy + /// endpoints want this high. #[serde(skip_serializing_if = "Option::is_none")] #[usage( name = "quic-max-streams", @@ -600,7 +577,7 @@ impl Config { keep_alive, mtu_discovery: self.mtu_discovery.unwrap_or(false), congestion_control: self.congestion_control, - receive_window: self.receive_window, + receive_window: self.receive_window.unwrap_or(DEFAULT_RECEIVE_WINDOW), stream_receive_window: self.stream_receive_window, send_window: self.send_window, qlog: self.qlog.clone(), @@ -627,15 +604,20 @@ fn validate_idle_timeout(idle_timeout: Option) -> crate::Result<()> { /// The largest value a QUIC varint can carry, which bounds every flow-control /// limit that goes on the wire as a transport parameter. +#[cfg(feature = "_transport")] const MAX_VARINT: u64 = (1 << 62) - 1; /// The largest stream limit QUIC can advertise: a stream ID is a varint whose /// low two bits are its type, so RFC 9000 caps the count at 2^60. +#[cfg(feature = "_transport")] const MAX_STREAMS: u64 = 1 << 60; -/// Reject a stream limit a peer would refuse with FRAME_ENCODING_ERROR. +/// Reject a stream limit no connection can honor: zero lets the peer open +/// nothing, and past 2^60 a peer refuses it with FRAME_ENCODING_ERROR. +#[cfg(feature = "_transport")] fn validate_streams(max_streams: Option) -> crate::Result<()> { match max_streams { + Some(0) => Err(crate::Error::StreamsRange), Some(streams) if streams > MAX_STREAMS => Err(crate::Error::StreamsRange), _ => Ok(()), } @@ -694,8 +676,11 @@ pub struct Resolved { pub mtu_discovery: bool, /// Congestion control override, or `None` for the default. pub congestion_control: Option, - /// Connection-wide receive window in bytes, or `None` for the 64 MiB default. - pub receive_window: Option, + /// Connection-wide receive window in bytes, defaulting to [`DEFAULT_RECEIVE_WINDOW`]. + /// + /// Resolved here rather than by each backend, whose own defaults differ (moq-quic's + /// and iroh's are unlimited), so every runtime honors the same `[quic]` section. + pub receive_window: u64, /// Per-stream receive window in bytes, or `None` for the backend default. pub stream_receive_window: Option, /// Cap on unacknowledged outgoing data in bytes, or `None` for the backend default. @@ -934,12 +919,12 @@ mod tests { assert!(at_limit.validate().is_ok()); } - /// Each window reaches `Resolved` untouched, and stays `None` when unset so each - /// backend applies the default itself (and the io_uring workers can tell it was unset). + /// An unset connection window resolves to our default, so every backend applies the + /// same one; the other windows stay `None` for the backend default. #[test] fn windows_resolve_per_field() { let unset = Config::default().resolve(); - assert_eq!(unset.receive_window, None); + assert_eq!(unset.receive_window, DEFAULT_RECEIVE_WINDOW); assert_eq!(unset.stream_receive_window, None); assert_eq!(unset.send_window, None); @@ -950,7 +935,7 @@ mod tests { ..Default::default() } .resolve(); - assert_eq!(set.receive_window, Some(64 << 20)); + assert_eq!(set.receive_window, 64 << 20); assert_eq!(set.stream_receive_window, Some(8 << 20)); assert_eq!(set.send_window, Some(32 << 20)); } @@ -1020,36 +1005,28 @@ mod tests { assert!(at_limit.validate().is_ok()); } - /// A MAX_STREAMS past 2^60 is a frame encoding error at the peer, so `[quic]` - /// and the runtime limits both refuse it. + /// A MAX_STREAMS past 2^60 is a frame encoding error at the peer, and zero lets + /// the peer open nothing, so `[quic]` refuses both. #[cfg(feature = "_transport")] #[test] fn stream_limits_must_fit_a_stream_id() { - let over = Config { - max_streams: Some(MAX_STREAMS + 1), - ..Default::default() - }; - assert!(matches!(over.validate(), Err(crate::Error::StreamsRange))); - let at_limit = Config { - max_streams: Some(MAX_STREAMS), - ..Default::default() - }; - assert!(at_limit.validate().is_ok()); - - let limit = |max_streams: u64, receive_window: u64| Limits { - max_streams: std::num::NonZeroU64::new(max_streams).unwrap(), - receive_window: std::num::NonZeroU64::new(receive_window).unwrap(), - send_window: std::num::NonZeroU64::MAX, - }; - assert!(limit(MAX_STREAMS, MAX_VARINT).validate().is_ok()); - assert!(matches!( - limit(MAX_STREAMS + 1, MAX_VARINT).validate(), - Err(crate::Error::StreamsRange) - )); - assert!(matches!( - limit(MAX_STREAMS, MAX_VARINT + 1).validate(), - Err(crate::Error::WindowRange("receive_window")) - )); + for max_streams in [0, MAX_STREAMS + 1] { + let over = Config { + max_streams: Some(max_streams), + ..Default::default() + }; + assert!( + matches!(over.validate(), Err(crate::Error::StreamsRange)), + "{max_streams}" + ); + } + for max_streams in [1, MAX_STREAMS] { + let at_limit = Config { + max_streams: Some(max_streams), + ..Default::default() + }; + assert!(at_limit.validate().is_ok(), "{max_streams}"); + } } #[test] diff --git a/rs/moq-tokio/src/quic/backend.rs b/rs/moq-tokio/src/quic/backend.rs index 1dbb82af289..b9a96ed3cc6 100644 --- a/rs/moq-tokio/src/quic/backend.rs +++ b/rs/moq-tokio/src/quic/backend.rs @@ -73,13 +73,12 @@ fn apply_transport(transport: &mut driver::TransportConfig, quic: &Resolved) { transport.congestion_controller_factory(congestion_factory(quic.congestion())); } -/// Apply the flow-control windows. The connection window defaults to ours, since the +/// Apply the flow-control windows. The connection window is always ours, since the /// backend's is unlimited; the others stay at the backend default when unset. fn apply_windows(transport: &mut driver::TransportConfig, quic: &Resolved) { // Saturating rather than erroring: `Config::validate` already rejects a window // past the varint, so this only bites a `Resolved` built without it. - let window = quic.receive_window.unwrap_or(crate::quic::DEFAULT_RECEIVE_WINDOW); - transport.receive_window(driver::VarInt::from_u64(window).unwrap_or(driver::VarInt::MAX)); + transport.receive_window(driver::VarInt::from_u64(quic.receive_window).unwrap_or(driver::VarInt::MAX)); if let Some(window) = quic.stream_receive_window { transport.stream_receive_window(driver::VarInt::from_u64(window).unwrap_or(driver::VarInt::MAX)); } @@ -96,6 +95,22 @@ fn congestion_factory(family: CongestionControl) -> Arc Self { + Self(conn) + } + + /// Change the flow-control limits `limits` sets; a raise reaches the peer on the next packet. + /// + /// Errors, changing nothing, on limits [`Limits::validate`](crate::quic::Limits::validate) refuses. + pub fn set_limits(&self, limits: crate::quic::Limits) -> Result<()> { + Ok(self.0.set_limits(limits)?) + } +} + /// Errors specific to the QUIC backend. #[derive(Debug, thiserror::Error)] #[non_exhaustive] @@ -253,6 +268,10 @@ pub enum Error { /// timeout. A host with a single address reports that error directly instead. #[error("all {} connection attempts failed: {}", .0.len(), crate::failover::describe(.0))] Failover(Vec>), + + /// The flow-control limits are ones a peer would refuse, so none were applied. + #[error("invalid quic limits: {0}")] + Limits(#[from] moq_quic::LimitsError), } impl crate::failover::Aggregate for Error { @@ -484,7 +503,8 @@ impl Error { | Self::InvalidDnsName | Self::NoRuntime | Self::NoInitialCipherSuite - | Self::ConnectRejected(_) => true, + | Self::ConnectRejected(_) + | Self::Limits(_) => true, Self::Tls(err) => err.is_terminal(), Self::Failover(failures) => { !failures.is_empty() && failures.iter().all(|failure| failure.error.is_terminal()) diff --git a/rs/moq-tokio/src/quic/driver/connection.rs b/rs/moq-tokio/src/quic/driver/connection.rs index 8de19c76139..a47922f60b5 100644 --- a/rs/moq-tokio/src/quic/driver/connection.rs +++ b/rs/moq-tokio/src/quic/driver/connection.rs @@ -794,6 +794,15 @@ impl Connection { // May need to send MAX_STREAMS to make progress conn.wake(); } + + /// See [`moq_quic::Connection::set_limits`] + pub fn set_limits(&self, limits: moq_quic::Limits) -> Result<(), moq_quic::LimitsError> { + let mut conn = self.0.state.lock("set_limits"); + conn.inner.set_limits(limits)?; + // May need to send MAX_DATA or MAX_STREAMS to make progress + conn.wake(); + Ok(()) + } } pin_project! { diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index f7f8107fd98..274f0c4fb59 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -708,7 +708,7 @@ impl Server { let Accepted { session, url, identity, authority, mut link } = within(deadline, crate::quic::accept(_conn, alpns)).await?; link.local = local; let binding = match url { Some(_) => moq_net::transport::Binding::Uri, None => moq_net::transport::Binding::InBand }; - let quic = Some(session.clone()); + let quic = Some(crate::quic::Connection::new((*session).clone())); let (request, deadline) = accept_request(&setup.server, session, binding, deadline).await?; // Only WebTransport carries a request URL; raw QUIC puts the path in the SETUP. let transport = match url { Some(_) => Transport::WebTransport, None => Transport::Quic }; @@ -1576,10 +1576,9 @@ pub struct Request { /// The link facts the transport could see. link: Link, kind: RequestKind, - /// The connection under a QUIC or WebTransport request, for - /// [`Request::set_quic_limits`]. + /// The connection under a QUIC or WebTransport request, for [`Request::quic`]. #[cfg(feature = "quic")] - quic: Option, + quic: Option, /// What remains of the handshake deadline, which also bounds [`Request::ok`]. deadline: Option, } @@ -1683,29 +1682,11 @@ impl Request { self } - /// Change this connection's QUIC flow-control limits, such as widening them - /// once the peer is known to be another relay. - /// - /// Errors on a transport that is not QUIC (qmux streams, iroh), which keeps - /// the limits its own config set, and on limits [`Limits::validate`](crate::quic::Limits::validate) - /// refuses, leaving the limits unchanged. - pub fn set_quic_limits(&self, limits: crate::quic::Limits) -> crate::Result<()> { - #[cfg(feature = "quic")] - if let Some(conn) = &self.quic { - use crate::quic::driver::VarInt; - limits.validate()?; - let window = VarInt::from_u64(limits.receive_window.get()).expect("validated"); - // Windows first: each setter can flush on its own, and the stream credit is - // what a peer waits on, so it must not overtake the window. - conn.set_receive_window(window); - conn.set_send_window(limits.send_window.get()); - let streams = VarInt::from_u64(limits.max_streams.get()).expect("validated"); - conn.set_max_concurrent_bi_streams(streams); - conn.set_max_concurrent_uni_streams(streams); - return Ok(()); - } - let _ = limits; - Err(Error::QuicLimits(self.transport)) + /// The QUIC connection under this request, to change its limits before answering + /// SETUP. `None` on a transport moq-quic does not drive (qmux streams, iroh). + #[cfg(feature = "quic")] + pub fn quic(&self) -> Option<&crate::quic::Connection> { + self.quic.as_ref() } /// Accept the session and return its supervised driver task. diff --git a/rs/moq-uring/src/error.rs b/rs/moq-uring/src/error.rs index 5a871aa995b..b4a43aa6631 100644 --- a/rs/moq-uring/src/error.rs +++ b/rs/moq-uring/src/error.rs @@ -11,14 +11,6 @@ pub enum Error { /// An io_uring or socket operation failed. #[error("io error: {0}")] Io(#[from] std::io::Error), - - /// A QUIC flow-control window does not fit a QUIC varint; names which one. - #[error("quic {0} must fit a QUIC varint")] - WindowRange(&'static str), - - /// A QUIC stream limit is past the 2^60 a peer accepts. - #[error("quic max_streams must be at most 2^60")] - StreamsRange, } impl Error { diff --git a/rs/moq-uring/src/quic/driver/connection.rs b/rs/moq-uring/src/quic/driver/connection.rs index 3f8f063eaae..2b8fe57e020 100644 --- a/rs/moq-uring/src/quic/driver/connection.rs +++ b/rs/moq-uring/src/quic/driver/connection.rs @@ -372,25 +372,13 @@ impl Connection { &self.handshake } - /// Change the connection's flow-control limits; a raise reaches the peer + /// Change the flow-control limits `limits` sets; a raise reaches the peer /// on the next packet. /// - /// Errors, changing nothing, on a stream limit past 2^60 or a receive window - /// past a QUIC varint, which a peer would refuse. - pub fn set_limits(&self, limits: crate::quic::Limits) -> Result<(), crate::Error> { - // A stream ID is a varint whose low two bits are its type (RFC 9000). - if limits.max_streams.get() > 1 << 60 { - return Err(crate::Error::StreamsRange); - } - let streams = VarInt::from_u64(limits.max_streams.get()).expect("checked above"); - let window = - VarInt::from_u64(limits.receive_window.get()).map_err(|_| crate::Error::WindowRange("receive_window"))?; - let mut conn = self.shared.conn.borrow_mut(); - conn.set_max_concurrent_streams(Dir::Bi, streams); - conn.set_max_concurrent_streams(Dir::Uni, streams); - conn.set_receive_window(window); - conn.set_send_window(limits.send_window.get()); - drop(conn); + /// Errors, changing nothing, on limits [`Limits::validate`](crate::quic::Limits::validate) + /// refuses. + pub fn set_limits(&self, limits: crate::quic::Limits) -> Result<(), Error> { + self.shared.conn.borrow_mut().set_limits(limits)?; self.shared.kick(); Ok(()) } diff --git a/rs/moq-uring/src/quic/mod.rs b/rs/moq-uring/src/quic/mod.rs index 32cfb5908be..2e30ef674e3 100644 --- a/rs/moq-uring/src/quic/mod.rs +++ b/rs/moq-uring/src/quic/mod.rs @@ -32,6 +32,8 @@ mod driver; pub use driver::{Connection, Credit, RecvStream, SendGroup, SendStream}; pub use endpoint::Endpoint; +/// The flow-control limits a live connection can change; see [`Connection::set_limits`]. +pub use moq_quic::{Limits, LimitsError}; /// The QUIC payload size every full datagram in a GSO train uses, and the /// stride GRO coalesces with. @@ -161,22 +163,6 @@ impl Default for Transport { } } -/// The flow-control limits an established connection can change, such as a -/// relay widening them once it knows the peer is another relay. -/// -/// [`Connection::set_limits`] refuses a stream count past 2^60 or a receive -/// window past a QUIC varint. Each is non-zero because zero streams or a zero -/// window credits nothing and wedges the connection. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub struct Limits { - /// Max concurrent streams the peer may open, per direction. - pub max_streams: std::num::NonZeroU64, - /// Connection-wide receive window, in bytes. - pub receive_window: std::num::NonZeroU64, - /// Cap on unacknowledged outgoing data, in bytes. - pub send_window: std::num::NonZeroU64, -} - /// The congestion control family a connection runs. /// /// CUBIC is used for loss-based control and BBRv3 for delay-based control. @@ -264,6 +250,9 @@ pub enum Error { /// This endpoint closed the connection locally. #[error("locally closed")] Closed, + /// The flow-control limits are ones a peer would refuse, so none were applied. + #[error("invalid quic limits: {0}")] + Limits(#[from] LimitsError), } impl web_transport_trait::Error for Error { diff --git a/rs/moq-uring/src/quic/web.rs b/rs/moq-uring/src/quic/web.rs index b688ece4c32..97e4cac3726 100644 --- a/rs/moq-uring/src/quic/web.rs +++ b/rs/moq-uring/src/quic/web.rs @@ -483,7 +483,7 @@ impl Session { /// Change the underlying connection's flow-control limits; see /// [`Connection::set_limits`]. - pub fn set_limits(&self, limits: super::Limits) -> Result<(), crate::Error> { + pub fn set_limits(&self, limits: super::Limits) -> Result<(), super::Error> { self.conn.set_limits(limits) } From 710e7796e71141dae1c00c0bc6a0f7784763ed64 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 10:02:57 -0700 Subject: [PATCH 2/2] fix(quic): resolve every window default once, for both runtimes `quic::Config::resolve` now fills a 4 MiB per-stream receive window and a 16 MiB send window alongside the 64 MiB connection window, so the tokio, iroh, and io_uring backends run identical windows for the same `[quic]` section instead of each falling back to its own. Co-Authored-By: Claude Opus 5.5 --- doc/bin/relay/config.md | 4 +-- rs/moq-relay/src/uring.rs | 48 ++++++++++++++++++++++---------- rs/moq-tokio/src/iroh.rs | 10 ++----- rs/moq-tokio/src/quic.rs | 42 +++++++++++++++++----------- rs/moq-tokio/src/quic/backend.rs | 33 ++++++++++------------ 5 files changed, 79 insertions(+), 58 deletions(-) diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 4fdb821c24b..1c206ab141e 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -62,8 +62,8 @@ keep_alive = "3s" # Ping interval; "0s" disables it. iroh sti 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. ``` diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 2f827eafbe3..bb0ff7a6c31 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -423,13 +423,11 @@ fn transport(quic: &moq_tokio::quic::Resolved) -> anyhow::Result moq_uring::quic::Congestion::Loss, // Unset means the backend's own default, and live media wants a steady @@ -879,15 +877,37 @@ mod tests { assert_eq!(transport.send_window, Some(32 << 20)); } - /// Leaving the windows unset is the ordinary case. The connection window is - /// `[quic]`'s documented default, the same the tokio workers run; the - /// worker keeps its own per-stream and send credits. + /// The same `[quic]` section runs the same windows on both runtimes, set or + /// left at the documented defaults: the tokio backends apply `Resolved`'s + /// windows as they are, and the workers must get exactly those. #[test] - fn defaults_are_accepted() { - let transport = transport(&moq_tokio::quic::Config::default().resolve()).expect("defaults must build"); - assert_eq!(transport.receive_window, Some(moq_tokio::quic::DEFAULT_RECEIVE_WINDOW)); - assert_eq!(transport.stream_receive_window, None); - assert_eq!(transport.send_window, None); + #[allow(clippy::field_reassign_with_default)] + fn windows_match_the_tokio_runtime() { + let mut set = moq_tokio::quic::Config::default(); + set.receive_window = Some(128 << 20); + set.stream_receive_window = Some(8 << 20); + set.send_window = Some(32 << 20); + + for (quic, expected) in [ + (moq_tokio::quic::Config::default(), (64 << 20, 4 << 20, 16 << 20)), + (set, (128 << 20, 8 << 20, 32 << 20)), + ] { + let resolved = quic.resolve(); + let tokio = ( + resolved.receive_window, + resolved.stream_receive_window, + resolved.send_window, + ); + assert_eq!(tokio, expected); + + let transport = transport(&resolved).expect("windows must be accepted"); + let uring = ( + transport.receive_window, + transport.stream_receive_window, + transport.send_window, + ); + assert_eq!(uring, (Some(tokio.0), Some(tokio.1), Some(tokio.2))); + } } /// An auth decision still pending at the drain deadline is abandoned then, diff --git a/rs/moq-tokio/src/iroh.rs b/rs/moq-tokio/src/iroh.rs index b4256ad7722..74ae2011d6c 100644 --- a/rs/moq-tokio/src/iroh.rs +++ b/rs/moq-tokio/src/iroh.rs @@ -290,13 +290,9 @@ fn transport_config(quic: &crate::quic::Resolved) -> iroh::endpoint::QuicTranspo } let window = iroh::endpoint::VarInt::from_u64(quic.receive_window).unwrap_or(iroh::endpoint::VarInt::MAX); transport = transport.receive_window(window); - if let Some(window) = quic.stream_receive_window { - let window = iroh::endpoint::VarInt::from_u64(window).unwrap_or(iroh::endpoint::VarInt::MAX); - transport = transport.stream_receive_window(window); - } - if let Some(window) = quic.send_window { - transport = transport.send_window(window); - } + let window = iroh::endpoint::VarInt::from_u64(quic.stream_receive_window).unwrap_or(iroh::endpoint::VarInt::MAX); + transport = transport.stream_receive_window(window); + transport = transport.send_window(quic.send_window); transport.congestion_controller_factory(congestion_factory(quic.congestion())) } diff --git a/rs/moq-tokio/src/quic.rs b/rs/moq-tokio/src/quic.rs index ddab67eb398..43e91c42de5 100644 --- a/rs/moq-tokio/src/quic.rs +++ b/rs/moq-tokio/src/quic.rs @@ -115,6 +115,16 @@ pub const DEFAULT_MAX_STREAMS: u64 = 10_000; /// shares one connection, can still sustain several Gbps at intercontinental RTTs. pub const DEFAULT_RECEIVE_WINDOW: u64 = 64 << 20; +/// Default per-stream receive window, used when [`Config::stream_receive_window`] is unset. +/// +/// A quarter of [`DEFAULT_RECEIVE_WINDOW`], so one slow group can't take the whole +/// connection, and enough for a single group stream to reach a few hundred Mbps at a +/// 100 ms RTT. +pub const DEFAULT_STREAM_RECEIVE_WINDOW: u64 = 4 << 20; + +/// Default cap on unacknowledged outgoing data, used when [`Config::send_window`] is unset. +pub const DEFAULT_SEND_WINDOW: u64 = 16 << 20; + /// Default idle timeout before an inactive connection is dropped. /// /// A peer that dies or drops off the network sends no close, so this is how long @@ -248,7 +258,7 @@ pub struct Config { )] pub receive_window: Option, - /// Per-stream receive window, in bytes. Unset leaves the backend default. + /// Per-stream receive window, in bytes. Defaults to 4 MiB. /// /// Keep it below [`receive_window`](Self::receive_window) so one slow group /// can't starve the rest of the connection. @@ -262,7 +272,7 @@ pub struct Config { pub stream_receive_window: Option, /// Cap on unacknowledged outgoing data, in bytes, regardless of what the peer - /// allows. Unset leaves the backend default. + /// allows. Defaults to 16 MiB. /// /// This bounds the transport send buffer. #[serde(skip_serializing_if = "Option::is_none")] @@ -578,8 +588,8 @@ impl Config { mtu_discovery: self.mtu_discovery.unwrap_or(false), congestion_control: self.congestion_control, receive_window: self.receive_window.unwrap_or(DEFAULT_RECEIVE_WINDOW), - stream_receive_window: self.stream_receive_window, - send_window: self.send_window, + stream_receive_window: self.stream_receive_window.unwrap_or(DEFAULT_STREAM_RECEIVE_WINDOW), + send_window: self.send_window.unwrap_or(DEFAULT_SEND_WINDOW), qlog: self.qlog.clone(), } } @@ -678,13 +688,14 @@ pub struct Resolved { pub congestion_control: Option, /// Connection-wide receive window in bytes, defaulting to [`DEFAULT_RECEIVE_WINDOW`]. /// - /// Resolved here rather than by each backend, whose own defaults differ (moq-quic's - /// and iroh's are unlimited), so every runtime honors the same `[quic]` section. + /// The windows are resolved here rather than by each backend, whose own defaults + /// differ (moq-quic's and iroh's connection window is unlimited), so every runtime + /// honors the same `[quic]` section. pub receive_window: u64, - /// Per-stream receive window in bytes, or `None` for the backend default. - pub stream_receive_window: Option, - /// Cap on unacknowledged outgoing data in bytes, or `None` for the backend default. - pub send_window: Option, + /// Per-stream receive window in bytes, defaulting to [`DEFAULT_STREAM_RECEIVE_WINDOW`]. + pub stream_receive_window: u64, + /// Cap on unacknowledged outgoing data in bytes, defaulting to [`DEFAULT_SEND_WINDOW`]. + pub send_window: u64, /// Directory to write qlog traces into, or `None` to not capture them. pub qlog: Option, } @@ -919,14 +930,13 @@ mod tests { assert!(at_limit.validate().is_ok()); } - /// An unset connection window resolves to our default, so every backend applies the - /// same one; the other windows stay `None` for the backend default. + /// An unset window resolves to our default, so every backend applies the same one. #[test] fn windows_resolve_per_field() { let unset = Config::default().resolve(); assert_eq!(unset.receive_window, DEFAULT_RECEIVE_WINDOW); - assert_eq!(unset.stream_receive_window, None); - assert_eq!(unset.send_window, None); + assert_eq!(unset.stream_receive_window, DEFAULT_STREAM_RECEIVE_WINDOW); + assert_eq!(unset.send_window, DEFAULT_SEND_WINDOW); let set = Config { receive_window: Some(64 << 20), @@ -936,8 +946,8 @@ mod tests { } .resolve(); assert_eq!(set.receive_window, 64 << 20); - assert_eq!(set.stream_receive_window, Some(8 << 20)); - assert_eq!(set.send_window, Some(32 << 20)); + assert_eq!(set.stream_receive_window, 8 << 20); + assert_eq!(set.send_window, 32 << 20); } /// A zero window credits nothing, so it would wedge the connection rather than diff --git a/rs/moq-tokio/src/quic/backend.rs b/rs/moq-tokio/src/quic/backend.rs index b9a96ed3cc6..0c7573d3fbb 100644 --- a/rs/moq-tokio/src/quic/backend.rs +++ b/rs/moq-tokio/src/quic/backend.rs @@ -73,18 +73,14 @@ fn apply_transport(transport: &mut driver::TransportConfig, quic: &Resolved) { transport.congestion_controller_factory(congestion_factory(quic.congestion())); } -/// Apply the flow-control windows. The connection window is always ours, since the -/// backend's is unlimited; the others stay at the backend default when unset. +/// Apply the flow-control windows `Config::resolve` settled, replacing moq-quic's own. fn apply_windows(transport: &mut driver::TransportConfig, quic: &Resolved) { // Saturating rather than erroring: `Config::validate` already rejects a window // past the varint, so this only bites a `Resolved` built without it. transport.receive_window(driver::VarInt::from_u64(quic.receive_window).unwrap_or(driver::VarInt::MAX)); - if let Some(window) = quic.stream_receive_window { - transport.stream_receive_window(driver::VarInt::from_u64(window).unwrap_or(driver::VarInt::MAX)); - } - if let Some(window) = quic.send_window { - transport.send_window(window); - } + transport + .stream_receive_window(driver::VarInt::from_u64(quic.stream_receive_window).unwrap_or(driver::VarInt::MAX)); + transport.send_window(quic.send_window); } /// The controller factory for a congestion control family. moq-quic's BBR is v3. @@ -1082,9 +1078,8 @@ mod tests { } /// moq-quic exposes no getters for the flow-control windows, but its `Debug` prints - /// them, which is enough to prove each one reached the transport config, that an - /// unset connection window gets our finite default instead of moq-quic's unlimited one, - /// and that the other unset knobs leave moq-quic's own defaults in place. + /// them, which is enough to prove each one reached the transport config, and that + /// unset windows get our defaults instead of moq-quic's. #[test] fn apply_windows_writes_each_field() { let quic = crate::quic::Config { @@ -1108,14 +1103,14 @@ mod tests { let mut untouched = driver::TransportConfig::default(); apply_windows(&mut untouched, &crate::quic::Config::default().resolve()); - let mut expected = driver::TransportConfig::default(); - expected.receive_window(driver::VarInt::from_u64(crate::quic::DEFAULT_RECEIVE_WINDOW).unwrap()); - assert_eq!(format!("{:?}", untouched), format!("{:?}", expected)); - assert_ne!( - format!("{:?}", untouched), - format!("{:?}", driver::TransportConfig::default()), - "moq-quic's own connection window is unlimited, so ours must replace it" - ); + let applied = format!("{:?}", untouched); + for field in [ + format!("receive_window: {}", crate::quic::DEFAULT_RECEIVE_WINDOW), + format!("stream_receive_window: {}", crate::quic::DEFAULT_STREAM_RECEIVE_WINDOW), + format!("send_window: {}", crate::quic::DEFAULT_SEND_WINDOW), + ] { + assert!(applied.contains(&field), "{field} missing from {applied}"); + } } /// Build a controller from each family's factory and downcast it to the