From 2645bcc342ee77c49308b6367f467573866bf3c9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 3 Oct 2026 09:56:50 -0700 Subject: [PATCH] fix(moq-tokio): finite default connection receive window (noq reassembly cap, part 1) (#4605) Co-authored-by: Claude Opus 5.5 --- doc/bin/relay/config.md | 10 +++++++--- doc/lib/rs/index.md | 3 ++- rs/moq-tokio/src/iroh.rs | 8 ++++---- rs/moq-tokio/src/noq.rs | 28 +++++++++++++++++----------- rs/moq-tokio/src/quic.rs | 16 ++++++++++++---- 5 files changed, 42 insertions(+), 23 deletions(-) diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 841d8bb13b..0524a10e71 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -55,9 +55,9 @@ idle_timeout = "10s" # Drop a connection after this long with no keep_alive = "3s" # Ping interval; "0s" disables it. Ignored by iroh. Default. gso = true # UDP segmentation offload. iroh cannot turn it off. mtu_discovery = false # Path MTU discovery. Default. -receive_window = 67108864 # Flow-control windows, in bytes. Omit for the backend default. -stream_receive_window = 8388608 -send_window = 33554432 +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. qlog = "/var/log/moq/qlog" # Existing directory. Needs the `qlog` build feature. ``` @@ -74,6 +74,10 @@ again before the deadline. Raise the receive windows when a fat, long path idles below the link rate: a window under the bandwidth-delay product stalls the sender waiting for credit. +`receive_window` bounds how much unread data one peer can make the relay buffer +across all of its streams; the transport's own default is unlimited, so it +defaults to 64 MiB, enough for a relay-to-relay session to carry several Gbps at +a 100 ms RTT. Keep `stream_receive_window` well under `receive_window` so one slow group cannot starve the connection. `send_window` caps unacknowledged outgoing data whatever the peer allows, bounding the transport send buffer. A zero window is refused, and the receive windows must fit a QUIC diff --git a/doc/lib/rs/index.md b/doc/lib/rs/index.md index ac11e13127..4170a70ef6 100644 --- a/doc/lib/rs/index.md +++ b/doc/lib/rs/index.md @@ -102,7 +102,8 @@ quic.send_window = Some(32 << 20); // unacknowledged data we may hold let client = moq_tokio::connect::Config::default().init(quic)?; ``` -Unset flow-control windows keep the transport defaults, and `init` errors on a +Unset flow-control windows keep the transport defaults, except `receive_window`, +which defaults to 64 MiB because the transport's is unlimited. `init` errors on a knob the transport cannot honor rather than dropping it: iroh cannot disable GSO. `quic::Resolved::default()` is what an untouched config resolves to, so read the defaults from there. The diff --git a/rs/moq-tokio/src/iroh.rs b/rs/moq-tokio/src/iroh.rs index 951260a137..43cfb02c2c 100644 --- a/rs/moq-tokio/src/iroh.rs +++ b/rs/moq-tokio/src/iroh.rs @@ -253,10 +253,10 @@ impl Config { if !quic.mtu_discovery { transport = transport.mtu_discovery_config(None); } - if let Some(window) = quic.receive_window { - let window = iroh::endpoint::VarInt::from_u64(window).unwrap_or(iroh::endpoint::VarInt::MAX); - transport = transport.receive_window(window); - } + // 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); + 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); diff --git a/rs/moq-tokio/src/noq.rs b/rs/moq-tokio/src/noq.rs index 15e3692d75..c831288691 100644 --- a/rs/moq-tokio/src/noq.rs +++ b/rs/moq-tokio/src/noq.rs @@ -60,13 +60,13 @@ fn apply_transport(transport: &mut noq::TransportConfig, quic: &Resolved) { transport.congestion_controller_factory(congestion_factory(quic.congestion())); } -/// Apply the flow-control windows, leaving each at the backend default when unset. +/// Apply the flow-control windows. The connection window defaults to ours, since the +/// backend's is unlimited; the others stay at the backend default when unset. fn apply_windows(transport: &mut noq::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. - if let Some(window) = quic.receive_window { - transport.receive_window(noq::VarInt::from_u64(window).unwrap_or(noq::VarInt::MAX)); - } + let window = quic.receive_window.unwrap_or(crate::quic::DEFAULT_RECEIVE_WINDOW); + transport.receive_window(noq::VarInt::from_u64(window).unwrap_or(noq::VarInt::MAX)); if let Some(window) = quic.stream_receive_window { transport.stream_receive_window(noq::VarInt::from_u64(window).unwrap_or(noq::VarInt::MAX)); } @@ -1043,14 +1043,13 @@ mod tests { } /// noq exposes no getters for the flow-control windows, but its `Debug` prints - /// them, which is enough to prove each one reached the transport config and that - /// an unset knob leaves noq's own default in place. + /// them, which is enough to prove each one reached the transport config, that an + /// unset connection window gets our finite default instead of noq's unlimited one, + /// and that the other unset knobs leave noq's own defaults in place. #[test] fn apply_windows_writes_each_field() { - let defaults = format!("{:?}", noq::TransportConfig::default()); - let mut quic = crate::quic::Config::default(); - quic.receive_window = Some(64 << 20); + quic.receive_window = Some(128 << 20); quic.stream_receive_window = Some(8 << 20); quic.send_window = Some(32 << 20); @@ -1059,7 +1058,7 @@ mod tests { let applied = format!("{:?}", transport); for field in [ - format!("receive_window: {}", 64 << 20), + format!("receive_window: {}", 128 << 20), format!("stream_receive_window: {}", 8 << 20), format!("send_window: {}", 32 << 20), ] { @@ -1068,7 +1067,14 @@ mod tests { let mut untouched = noq::TransportConfig::default(); apply_windows(&mut untouched, &crate::quic::Config::default().resolve()); - assert_eq!(format!("{:?}", untouched), defaults); + let mut expected = noq::TransportConfig::default(); + expected.receive_window(noq::VarInt::from_u64(crate::quic::DEFAULT_RECEIVE_WINDOW).unwrap()); + assert_eq!(format!("{:?}", untouched), format!("{:?}", expected)); + assert_ne!( + format!("{:?}", untouched), + format!("{:?}", noq::TransportConfig::default()), + "noq's own connection window is unlimited, so ours must replace it" + ); } /// Build a controller from each family's factory and downcast it to the diff --git a/rs/moq-tokio/src/quic.rs b/rs/moq-tokio/src/quic.rs index 275ba84d6e..b375f95bc4 100644 --- a/rs/moq-tokio/src/quic.rs +++ b/rs/moq-tokio/src/quic.rs @@ -82,6 +82,14 @@ impl std::str::FromStr for CongestionControl { /// Default maximum number of concurrent QUIC streams (bidi and uni) per connection. pub const DEFAULT_MAX_STREAMS: u64 = 10_000; +/// Default connection-wide receive window, used when [`Config::receive_window`] is unset. +/// +/// Bounds how much unread data one peer can make us buffer across all of its streams, +/// which the backends otherwise leave unlimited. Sized so a relay-to-relay session, which +/// shares one connection, can still sustain several Gbps at intercontinental RTTs. +#[cfg(any(feature = "noq", feature = "iroh"))] +pub(crate) const DEFAULT_RECEIVE_WINDOW: u64 = 64 << 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 @@ -189,7 +197,7 @@ pub struct Config { pub congestion_control: Option, /// Connection-wide receive window, in bytes: how much data the peer may have - /// in flight across every stream at once. Unset leaves the backend default. + /// in flight across every stream at once. Defaults to 64 MiB. /// /// Raise it when a fat, long path is idling below the link rate: the window /// has to cover the bandwidth-delay product or the peer stalls waiting for @@ -605,7 +613,7 @@ 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 backend default. + /// Connection-wide receive window in bytes, or `None` for the 64 MiB default. pub receive_window: Option, /// Per-stream receive window in bytes, or `None` for the backend default. pub stream_receive_window: Option, @@ -845,8 +853,8 @@ mod tests { assert!(at_limit.validate().is_ok()); } - /// Each window reaches `Resolved` untouched, and stays `None` when unset so the - /// backend keeps its own default rather than being pinned to ours. + /// 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). #[test] fn windows_resolve_per_field() { let unset = Config::default().resolve();