diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 8d42bf3687c..0913cd73091 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -330,7 +330,11 @@ moq --cluster-lan --cluster-lan-secret /etc/moq/cluster.key import capture MoQ process and `moq-relay` on the LAN, with no other configuration. Without `--cluster-lan-secret`, anyone who can reach the listener joins, so leave it unset only on networks you trust. `--cluster-connect` is a MoQ side on its own, -so it needs no `--connect`. +so it needs no `--connect`. WAN cluster dials (`--cluster-connect`, +`--cluster-connect-api`) share `--connect-bind` with `--connect`, and each binds +it, while LAN dials take an ephemeral port. So a fixed `--connect-bind` port with +both `--connect` and a WAN cluster peer is refused at startup: use port 0, or drop +`--connect` (unset `MOQ_CONNECT` if it supplies the URL). ## Auth diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 5faf7df0a11..8d4ddbc12cb 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -95,8 +95,21 @@ fetches = 10000 # Independent simultaneous FETCH requests. announces = 100000 subscriptions = 10000 fetches = 10000 + +[peer.quic] +max_streams = 100000 # Concurrent QUIC streams per direction. +receive_window = 671088640 # Connection-wide receive window (640 MiB). +send_window = 671088640 # Unacknowledged outgoing data (640 MiB). ``` +`[peer.quic]` takes only `max_streams`, `receive_window`, and `send_window`; +any other `[quic]` key is refused. They replace the `[quic]` values for a +peer's QUIC connection once its grant is known, before the SETUP reply. Links +the relay dials get them from the handshake on. A link carrying a whole cluster is then +not throttled by `[quic]` limits sized for one viewer, for example +`--peer-quic-max-streams=200000`. Peers over WebSocket, TCP, or iroh keep +their transport's own limits. + These are the defaults. For example, `--peer-subscriptions=50000` raises the subscription cap on cluster links. `--session-fetches` and `--peer-fetches` set the independent FETCH caps; concurrent callers of the same group each diff --git a/quest/m1/quic/README.md b/quest/m1/quic/README.md index 11c3c74df3c..4ca5ace6bb3 100644 --- a/quest/m1/quic/README.md +++ b/quest/m1/quic/README.md @@ -69,8 +69,6 @@ cache limits remain separate; bitrate pacing composes with this memory cap. tail grace - [Schedule datagrams by subscription priority](/quest/m1/quic/datagram-priority.md) - datagrams join their subscription's send group instead of preempting every stream -- [Relay peers get wider limits](/quest/m1/quic/peer-limits.md) - MAX_STREAMS - and MAX_DATA are raised after SETUP identifies a cluster peer - [A FIN below received data is a final size error](/quest/m1/quic/final-size-below-received.md) - a shrunken final size closes with FINAL_SIZE_ERROR instead of blocking the reader forever - [A stateless reset keeps a closed connection's reason](/quest/m1/quic/close-reason-after-reset.md) - a reset after a local transport error neither replaces the reason nor reports a second loss - [0-RTT streams adopt a raised stream window](/quest/m1/quic/zero-rtt-window-raise.md) - a server's larger stream window reaches the client's 0-RTT streams, so they no longer stall diff --git a/quest/m1/quic/peer-limits.md b/quest/m1/quic/peer-limits.md deleted file mode 100644 index 7a97758dab9..00000000000 --- a/quest/m1/quic/peer-limits.md +++ /dev/null @@ -1,46 +0,0 @@ -# [S] Relay peers get wider limits - -## Goal - -A relay-to-relay session runs with wider stream and data limits than a -viewer's session, decided after the handshake rather than at bind time. A -cluster peer carrying thousands of broadcasts is never throttled by the -concurrent-stream count sized for one browser, and the viewer-facing limits -stay small so one client cannot reserve a relay's memory. - -## Plan - -The in-tree `moq-quic` `Connection` (`rs/moq-quic/src/connection/mod.rs`) -already exposes runtime `set_max_concurrent_streams`, `set_receive_window`, -and `set_send_window`; a raise queues `MAX_STREAMS` and `MAX_DATA` on the next -packet, and a shrink is a debt paid as the peer consumes credit. The existing -setters provide this seam in `moq-quic`, which moq-tokio and moq-uring both -run on. - -Decided 2026-10-08: the relay's peer config surface already exists. A -session is a cluster peer when its admitted grant is marked `peer`, and the -`[peer]` table (beside `[session]` for ordinary clients) holds its request -caps; this quest extends that table with the QUIC values rather than adding -its own. - -- moq-net's transport trait gains `set_limits(Limits)` on the session, - `Limits` carrying the three values, implemented by the moq-tokio and - moq-uring adapters over `moq-quic`; every other - backend, the browser included, reports it unsupported. -- The peer table gains the same three values (the stream limit and the two - windows), - defaulting to an order of magnitude above the client defaults. `moq-relay` - applies it once a session is classified as a cluster peer, on the io_uring - workers too. -- Refuse a `peer` value below the client default rather than silently - shrinking. - -The [receive-memory plan](/quest/m1/quic/receive-memory.md) requires client -and peer memory allowances to fit the largest supported encoded unit, -including framing. Preserve that validation and retained-credit accounting -when applying runtime limits; wider stream counts must not create a second -unbounded assembly pool. - -Tests: a cluster session sees the raised `MAX_STREAMS` after SETUP and a -viewer session does not; the io_uring path applies the same values; a -`peer` table below the defaults is refused at resolve time. diff --git a/quest/m1/quic/receive-memory.md b/quest/m1/quic/receive-memory.md index da985e4eb66..476565e9437 100644 --- a/quest/m1/quic/receive-memory.md +++ b/quest/m1/quic/receive-memory.md @@ -107,4 +107,3 @@ separate guide quest is needed. - [Unbounded groups](/quest/m2/unbounded-groups/README.md) - removes the frame cap; frame payload credit becomes per segment - [Scope track priority](/quest/m1/track-priority-scope.md) - broadcast fairness domains and subscription priority semantics - [QUIC bitrate caps](/quest/m2/rate-quic.md) - pace credit grants inside the independently bounded memory allowance -- [Relay peers get wider limits](/quest/m1/quic/peer-limits.md) - client and peer windows must both fit supported units diff --git a/quest/m2/rate-quic.md b/quest/m2/rate-quic.md index 029b8414182..9e5c57c74d9 100644 --- a/quest/m2/rate-quic.md +++ b/quest/m2/rate-quic.md @@ -49,18 +49,20 @@ backpressure the peer cannot ignore, rather than metering and closing. window for an uncapped session, paced credit for a capped one. A lowered rate (revalidation, a union shrinking) withholds future grants against outstanding credit rather than shrinking assembly memory below a legal - message. Memory-window reductions retain the shrink-as-debt semantics - described in [peer limits](/quest/m1/quic/peer-limits.md). Overshoot is + message. Memory-window reductions retain `moq-quic`'s shrink-as-debt + semantics. Overshoot is bounded by credit outstanding at the change and the test asserts that bound. - **Egress: cap the pacer.** The send rate is `min(controller rate, cap)`, so a subscriber below the broadcast's bitrate gets MoQ's normal group skipping, not a growing queue. - **Runtime, per connection.** The cap is known only after auth (the token rides the CONNECT URL or arrives in band), so it is set on a live - connection and reset on revalidation or a union change. Reuse the runtime - limits seam [peer limits](/quest/m1/quic/peer-limits.md) adds - (`set_limits(Limits)` on moq-net's transport trait, implemented by the - moq-tokio and moq-uring adapters), extended with the two rates. + 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 + 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 uncapped grant lifts it, matching the union's permissive meaning. Decided @@ -89,5 +91,4 @@ that a one-second rate window can also be the complete assembly allowance. ## Required -- [Peer limits](/quest/m1/quic/peer-limits.md) - adds the runtime `set_limits(Limits)` seam and shrink-as-debt behavior this extends - [Bound message assembly](/quest/m1/quic/receive-memory.md) - separate retained memory from paced outstanding credit diff --git a/quest/m2/transport-adapter-dedup.md b/quest/m2/transport-adapter-dedup.md index 2e557dcdf2d..48f440b6244 100644 --- a/quest/m2/transport-adapter-dedup.md +++ b/quest/m2/transport-adapter-dedup.md @@ -27,4 +27,3 @@ Public API: possibly a feature-gated module. Wire: none. - [poll_acked on moq-net's send stream](/quest/m2/quic-ack-hook.md) - a backend-specific method the moq-tokio and moq-uring adapters keep - [Per-stream deadlines](/quest/m2/quic-deadline.md) - adds `set_deadline`, also kept out of the shared adapter - [QUIC caps](/quest/m2/rate-quic.md) - extends `set_limits` with the bitrate caps -- [Peer limits](/quest/m1/quic/peer-limits.md) - adds the `set_limits` seam diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index 14401ce2245..d3b36d93faf 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -28,7 +28,7 @@ doc = false [features] default = ["cluster-lan", "quic", "websocket", "nvidia", "s3", "gcs", "azure"] # Experimental transport with a heavy dependency tree, enabled only on request. -iroh = ["moq-tokio/iroh"] +iroh = ["moq-relay/iroh", "moq-tokio/iroh"] # LAN discovery and meshing (`--cluster-lan`). cluster-lan = ["moq-relay/cluster-lan"] jemalloc = ["moq-tokio/jemalloc"] diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index d89b157ba2b..e067c7cb6f3 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -550,6 +550,15 @@ impl MoqSide { self.client.url.is_some() || self.serves() || self.cluster_dials(), "a MoQ side is required: pass --connect to dial a relay, a --listen option to self-host, --cluster-lan to mesh over the LAN, or --cluster-connect to join a cluster" ); + // The `--connect` client and the cluster's WAN dials each bind + // `--connect-bind` for QUIC, so a fixed port would collide halfway through startup. + #[cfg(feature = "quic")] + anyhow::ensure!( + !(self.client.url.is_some() + && self.cluster_dials() + && self.client.bind.is_some_and(|bind| bind.port() != 0)), + "--connect-bind with a fixed port cannot serve both --connect and --cluster-connect / --cluster-connect-api, which each bind it: use port 0, or drop --connect (unset MOQ_CONNECT if it supplies the URL)" + ); #[cfg(feature = "cluster-lan")] { self.cluster.lan.validate()?; @@ -2315,6 +2324,51 @@ mod tests { parse(&[]).unserved().expect("nothing to refuse"); } + /// `--connect` and a WAN cluster peer each bind `--connect-bind` for QUIC, so a + /// fixed port is refused before anything binds; port 0 or either side alone is + /// fine, and a build without QUIC binds nothing to collide. + #[test] + fn a_fixed_connect_bind_serves_one_dialer() { + let parse = |flags: &[&str]| { + let args = ["moq"].iter().chain(flags).chain(&["import", "ts"]).copied(); + Invocation::try_parse_from(args).expect("parse") + }; + for peer in [ + ["--cluster-connect", "https://relay.example"], + ["--cluster-connect-api", "https://api.example/peers"], + ] { + let both = [ + &["--connect", "https://relay.example", "--connect-bind", "127.0.0.1:4443"][..], + &peer, + ] + .concat(); + let result = parse(&both).moq.validate(); + if cfg!(feature = "quic") { + let err = result.unwrap_err().to_string(); + assert!(err.contains("--connect-bind") && err.contains(peer[0]), "{err}"); + } else { + result.expect("without QUIC nothing binds the fixed port"); + } + + let ephemeral = [ + &["--connect", "https://relay.example", "--connect-bind", "127.0.0.1:0"][..], + &peer, + ] + .concat(); + parse(&ephemeral).moq.validate().expect("an ephemeral bind serves both"); + + let cluster_only = [&["--connect-bind", "127.0.0.1:4443"][..], &peer].concat(); + parse(&cluster_only) + .moq + .validate() + .expect("the cluster alone owns a fixed bind"); + } + parse(&["--connect", "https://relay.example", "--connect-bind", "127.0.0.1:4443"]) + .moq + .validate() + .expect("--connect alone owns a fixed bind"); + } + /// `--cluster-connect` / `--cluster-connect-api` attach the process as a /// cluster peer, so they are a MoQ side on their own. `--cluster-node` is /// identity, not an attachment. diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 6f6091c2505..9bdac6c4801 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -428,32 +428,50 @@ impl Directions { async fn spawn_moq( moq: &MoqSide, net: &Net, - client: moq_tokio::Client, cluster: moq_relay::cluster::Cluster, directions: Directions, tasks: &mut JoinSet>, ) -> anyhow::Result { - let mut bandwidth = moq_net::bandwidth::Allocator::unlimited(); - let mut connection = None; let cluster = cluster - .with_client(client.clone()) .with_client_tls(moq.client.tls.build()?) .with_connect(moq.client.clone(), moq.quic.clone()); + #[cfg(feature = "iroh")] + let cluster = match net.iroh.clone() { + Some(iroh) => cluster.with_iroh(iroh), + None => cluster, + }; let origin = cluster.origin.clone(); - if let Some(url) = moq.client.url.clone() { - let mut client = client; + // Built only to dial `--connect`: the cluster binds its own client from the + // same `[connect]` config, so a second one would contend for a fixed bind port. + // Built before the listener so a bad config fails before readiness, but dialed + // only after it: a later startup failure would drop a live dial unclosed. + let dialer = match moq.client.url.clone() { + Some(url) => Some((url, net.client(moq.client.clone())?)), + None => None, + }; + + let (started, shutdown) = + notify_when_initialized(spawn_server(tasks, moq, &cluster, net, directions), moq::notify_ready).await?; + // Even standalone: it is what reports a panic in the origin driver. + tasks.spawn(async move { started.run().await }); + + let mut bandwidth = moq_net::bandwidth::Allocator::unlimited(); + let mut connection = None; + let mut client = None; + if let Some((url, mut dialer)) = dialer { + client = Some(dialer.clone()); if directions.publish { - client = client.with_publisher(origin.consume()); + dialer = dialer.with_publisher(origin.consume()); } if directions.consume { - client = client.with_subscriber(origin.clone()); + dialer = dialer.with_subscriber(origin.clone()); } // Both directions ride one connection, so this dials directly rather than // through `Client::publish` / `Client::consume`, which each attach a single // direction and dial the same URL. - let reconnect = client.connect(url); + let reconnect = dialer.connect(url); // Read before the handle moves into the task. This consumer is persistent: it // survives reconnects, reading `None` while down, so it can be wired up before // anything connects. @@ -463,14 +481,10 @@ async fn spawn_moq( connection = Some(reconnect); } - let (started, shutdown) = - notify_when_initialized(spawn_server(tasks, moq, &cluster, net, directions), moq::notify_ready).await?; - // Even standalone: it is what reports a panic in the origin driver. - tasks.spawn(async move { started.run().await }); - Ok(Attached { bandwidth, origin, + client, connection, shutdown, }) @@ -480,6 +494,8 @@ async fn spawn_moq( struct Attached { bandwidth: moq_net::bandwidth::Allocator, origin: moq_net::origin::Producer, + /// The client that dialed `--connect`, to close its endpoint on exit. + client: Option, /// The relay connection, when `--connect` dialed one. connection: Option, shutdown: Option, @@ -513,10 +529,16 @@ async fn run_play(moq: MoqSide, args: play::Args, net: Net) -> anyhow::Result<() consume: true, ..Default::default() }; - let client = net.client(moq.client.clone())?; - let Attached { origin, .. } = spawn_moq(&moq, &net, client, cluster, directions, &mut tasks).await?; + let Attached { + origin, + client, + connection, + .. + } = spawn_moq(&moq, &net, cluster.clone(), directions, &mut tasks).await?; - play::run(origin.consume(), name, args, tasks).await + let result = play::run(origin.consume(), name, args, tasks).await; + detach(connection, client, &cluster).await; + result } /// Run every stage over one Origin and one MoQ attachment. @@ -537,15 +559,17 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res // The stage combinations were refused up front by `Invocation::validate`, before // anything bound a port or dialed out. - let client = net.client(moq.client.clone())?; + let mut client = None; let mut connection = None; let result = async { let Attached { bandwidth, origin, + client: dialer, connection: attached, shutdown, - } = spawn_moq(&moq, &net, client.clone(), cluster, Directions::of(&stages), &mut tasks).await?; + } = spawn_moq(&moq, &net, cluster.clone(), Directions::of(&stages), &mut tasks).await?; + client = dialer; connection = attached; // stdin and stdout are one resource each, so two stages can't share them. @@ -589,16 +613,29 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res } .await; - // The process exits next, even on a setup error, so the relay only hears we left - // if the close goes out now. The connection first delivers what it queued, such - // as the finished tracks at stdin EOF, since the client's close discards it. + // Even on a setup error. + detach(connection, client, &cluster).await; + result +} + +/// Close everything [`spawn_moq`] attached. The process exits next, so the relay +/// and cluster peers only hear we left if the closes go out now. The connection +/// first delivers what it queued, such as the finished tracks at stdin EOF, since +/// the client's close discards it. +async fn detach( + connection: Option, + client: Option, + cluster: &moq_relay::cluster::Cluster, +) { if let Some(connection) = connection && let Err(err) = connection.close().await { tracing::warn!(%err, "closed before delivering everything"); } - client.close().await; - result + if let Some(client) = client { + client.close().await; + } + cluster.close().await; } /// Run the non-Send pipelines on `local`, reporting each into `tasks`. @@ -1099,6 +1136,43 @@ mod tests { assert!(tasks.is_empty(), "nothing should be spawned after a bind failure"); } + /// `--connect` dials only once the listener is up, so a bind failure leaves no + /// live dial to drop without its close. + #[tokio::test] + async fn a_stream_bind_failure_starts_no_dial() { + let _ = moq_tokio::crypto::install_default(); + let occupied = std::net::TcpListener::bind("127.0.0.1:0").expect("occupy a port"); + let addr = occupied.local_addr().expect("occupied address").to_string(); + let invocation = Invocation::try_parse_from([ + "moq", + "--connect", + "http://127.0.0.1:9/", + "--listen-tcp-bind", + &addr, + "--auth-public", + "**", + "import", + "ts", + ]) + .expect("parse"); + let cluster = invocation.moq.cluster().expect("create cluster"); + let net = Net { + quic: invocation.moq.quic.clone(), + #[cfg(feature = "iroh")] + iroh: None, + }; + let mut tasks = JoinSet::new(); + let directions = Directions { + publish: true, + consume: false, + }; + let Err(err) = spawn_moq(&invocation.moq, &net, cluster, directions, &mut tasks).await else { + panic!("the occupied port must fail startup"); + }; + assert!(err.to_string().contains("failed to bind listeners"), "{err:#}"); + assert!(tasks.is_empty(), "no dial starts before the listener is up"); + } + /// An explicit listener with no auth source stops startup instead of refusing /// every session, even for a caller that skipped `MoqSide::validate`. #[tokio::test] @@ -1191,17 +1265,10 @@ mod tests { ]) .expect("parse"); assert!(invocation.moq.validate().is_ok()); - let net = Net { - quic: invocation.moq.quic.clone(), - #[cfg(feature = "iroh")] - iroh: None, - }; - let client = net.client(invocation.moq.client.clone()).expect("client"); let cluster = invocation .moq .cluster() .expect("cluster") - .with_client(client) .with_connect(invocation.moq.client.clone(), invocation.moq.quic.clone()); let err = cluster .clone() @@ -1218,6 +1285,51 @@ mod tests { .expect("http API with TLS"); } + /// Without `--connect`, only the cluster binds `--connect-bind`, so a fixed + /// port with a WAN cluster peer starts instead of colliding with itself. + #[tokio::test] + async fn a_fixed_connect_bind_with_a_cluster_peer_starts() { + let _ = moq_tokio::crypto::install_default(); + // Borrow a free port from the OS, then hand it to `--connect-bind`. + let bind = std::net::UdpSocket::bind("127.0.0.1:0") + .expect("probe socket") + .local_addr() + .expect("probe addr") + .to_string(); + let invocation = Invocation::try_parse_from([ + "moq", + "--connect-bind", + &bind, + "--cluster-connect", + "moql://127.0.0.1:9/", + "import", + "ts", + ]) + .expect("parse"); + invocation.moq.validate().expect("validate"); + let cluster = invocation.moq.cluster().expect("cluster"); + let net = Net { + quic: invocation.moq.quic.clone(), + #[cfg(feature = "iroh")] + iroh: None, + }; + let mut tasks = JoinSet::new(); + let attached = spawn_moq( + &invocation.moq, + &net, + cluster, + Directions { + publish: true, + consume: false, + }, + &mut tasks, + ) + .await + .expect("a fixed bind with a cluster peer starts"); + assert!(attached.client.is_none(), "no --connect, so no client of its own"); + tasks.abort_all(); + } + #[test] fn explicit_stream_listeners_are_public_without_exposing_mesh_quic() { assert!(is_public_transport(moq_tokio::Transport::Tcp, false)); diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index 11abe4ac78e..6014c310ac5 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -9,6 +9,7 @@ use std::{ }; use anyhow::Context; +use futures::FutureExt as _; use moq_net::origin; use moq_net::{Hop, stats::Tier}; use reqwest_middleware::ClientWithMiddleware; @@ -433,16 +434,27 @@ impl DialSources { } } +/// A dial task that ends once its [`DialEntry`] is replaced or the [`DialMap`] +/// drops, so a cancelled `run` takes the dynamic dials, spawned outside its +/// `JoinSet`, down with it. +struct DialTask(AbortHandle); + +impl Drop for DialTask { + fn drop(&mut self) { + self.0.abort(); + } +} + /// One entry in [`DialMap`]. struct DialEntry { - handle: AbortHandle, + task: DialTask, sources: DialSources, active: DialSource, } -/// Map of in-flight cluster dials, keyed by canonical peer identity. Cloneable: the inner -/// map is shared via `Arc>` so the discovery tasks and the static-seed -/// phase write to the same set of entries. +/// Map of in-flight cluster dials, keyed by canonical peer identity, owning each +/// dial's task. Cloneable: the inner map is shared via `Arc>` so the +/// discovery tasks and the static-seed phase write to the same set of entries. #[derive(Clone, Default)] struct DialMap { inner: Arc>>, @@ -479,20 +491,19 @@ impl DialMap { let replace = entry.active == source && entry.sources.get(source) != Some(&target); entry.sources.set(source, target.clone()); if replace { - entry.handle.abort(); - entry.handle = spawn(target); + entry.task = DialTask(spawn(target)); } return; } let key = target.key.clone(); - let handle = spawn(target.clone()); + let task = DialTask(spawn(target.clone())); let mut sources = DialSources::default(); sources.set(source, target); map.insert( key, DialEntry { - handle, + task, sources, active: source, }, @@ -520,14 +531,12 @@ impl DialMap { let next_target = next_target.clone(); entry.active = next_source; if active_target.as_ref() != Some(&next_target) { - entry.handle.abort(); - entry.handle = spawn(next_target); + entry.task = DialTask(spawn(next_target)); } return; } - let entry = map.remove(peer).expect("entry exists"); - entry.handle.abort(); + map.remove(peer); } /// Reconcile the API source against `desired`, including changes to a peer's @@ -554,6 +563,148 @@ impl DialMap { } } +/// Every dial task and QUIC client the cluster dials peers with, so +/// [`Cluster::close`] can end each link and send the peer its close before the +/// runtime stops: a dropped QUIC link only queues the close, and the peer then +/// keeps the link and its routes until its idle timeout. +#[derive(Clone)] +struct Dialers { + state: Arc>, + /// Set on close. Each dial task holds a receiver, so [`watch::Sender::closed`] + /// waits for every one of them to end. + /// + /// [`watch::Sender::closed`]: tokio::sync::watch::Sender::closed + stop: Arc>, +} + +struct DialersPhase { + /// `None` once closed, which refuses any further dial. + open: Option, + /// The shutdown the first close started, which every close awaits, so none + /// returns before it is done. + closing: Option>>, +} + +struct DialersState { + clients: HashMap, + next: u64, + /// Cloned into each background close, so `finished` ends once they all have. + closing: tokio::sync::mpsc::Sender<()>, + finished: tokio::sync::mpsc::Receiver<()>, +} + +impl Dialers { + fn new() -> Self { + let (closing, finished) = tokio::sync::mpsc::channel(1); + Self { + state: Arc::new(Mutex::new(DialersPhase { + open: Some(DialersState { + clients: HashMap::new(), + next: 0, + closing, + finished, + }), + closing: None, + })), + stop: Arc::new(tokio::sync::watch::Sender::new(false)), + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, DialersPhase> { + self.state.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Keep `client` until [`Self::close`]. + fn insert(&self, client: &moq_tokio::Client) -> anyhow::Result { + let mut phase = self.lock(); + let state = phase.open.as_mut().context("cluster closed")?; + let id = state.next; + state.next += 1; + state.clients.insert(id, client.clone()); + Ok(id) + } + + /// Keep `client` until [`Self::close`] or until the guard drops, which closes it + /// in the background. + fn hold(&self, client: &moq_tokio::Client) -> anyhow::Result { + Ok(Held { + id: self.insert(client)?, + dialers: self.clone(), + }) + } + + /// Run one dial task until it ends or [`Self::close`] stops it. + async fn supervise(&self, dial: impl std::future::Future) { + let mut stop = self.stop.subscribe(); + tokio::select! { + // First, so a dial spawned just before the close never starts. + biased; + _ = stop.wait_for(|stopped| *stopped) => {} + () = dial => {} + } + } + + /// Stop every dial task and close every client, including those a dropped + /// guard is still closing, and refuse further dials. + async fn close(&self) { + let closing = { + let mut phase = self.lock(); + let phase = &mut *phase; + phase + .closing + .get_or_insert_with(|| { + let state = phase.open.take().expect("open until the first close"); + Self::shutdown(state, self.stop.clone()).boxed().shared() + }) + .clone() + }; + closing.await; + } + + async fn shutdown(state: DialersState, stop: Arc>) { + stop.send_replace(true); + let DialersState { + clients, + closing, + mut finished, + .. + } = state; + futures::future::join_all(clients.into_values().map(moq_tokio::Client::close)).await; + drop(closing); + // Nothing is ever sent: this returns `None` once every sender has dropped. + let _ = finished.recv().await; + // Each stopped task dropped its link, which ends a non-QUIC one. + stop.closed().await; + } +} + +/// A dial's claim on its client in [`Dialers`]. +struct Held { + dialers: Dialers, + id: u64, +} + +impl Drop for Held { + fn drop(&mut self) { + let mut phase = self.dialers.lock(); + // Already taken by `Dialers::close`, which closes the client itself. + let Some(state) = phase.open.as_mut() else { + return; + }; + let Some(client) = state.clients.remove(&self.id) else { + return; + }; + let closing = state.closing.clone(); + // Without a runtime nothing could send the close anyway. + if let Ok(runtime) = tokio::runtime::Handle::try_current() { + runtime.spawn(async move { + client.close().await; + drop(closing); + }); + } + } +} + /// Configuration for relay clustering. /// /// [`Self::connect`] / [`Self::connect_api`] list peers to dial, and @@ -811,14 +962,14 @@ struct LanAuth { /// Construction settings for a [`Cluster`]: identity, discovery, and the origin cache. /// /// The origin is built once from these, so cache settings cannot detach a handle -/// taken after construction. Independent services ([`Cluster::with_client`], +/// taken after construction. Independent services ([`Cluster::with_connect`], /// [`Cluster::with_stats`]) attach afterwards without rebuilding it. #[derive(Default)] #[non_exhaustive] pub struct Options { /// Request caps for ordinary client sessions. pub session: crate::limits::Session, - /// Request caps for authenticated peers and outbound cluster links. + /// Request caps and QUIC limits for authenticated peers and outbound cluster links. pub peer: crate::limits::Peer, /// Cluster identity, peers, and discovery. pub config: Config, @@ -872,19 +1023,34 @@ impl Started { /// Run the cluster until it stops, or until a task it cannot serve without (its /// origin driver, cache governor, or certificate reload) panics. /// - /// Standalone, this only waits for such a panic. + /// Standalone, this only waits for such a panic. Returning, a failure + /// included, closes every dial as [`Cluster::close`] does. A cancelled `run` + /// starts each link's close but cannot wait for it, so a caller that must have + /// peers told before exiting awaits [`Cluster::close`] or lets `run` return. pub async fn run(self) -> anyhow::Result<()> { + // Registered only now, so a start that fails or never runs drops its client + // and frees the bind. A closed cluster has nothing left to run. + if let Some(dial) = &self.cluster.dial + && self.cluster.dialers.insert(dial).is_err() + { + return Ok(()); + } let critical = self.cluster.critical.clone(); + let dialers = self.cluster.dialers.clone(); let work = async move { match self.work { Some(work) => self.cluster.run_work(work).await, None => std::future::pending().await, } }; - tokio::select! { + let result = tokio::select! { res = work => res, err = critical.failed() => Err(err), - } + }; + // The dial tasks went with `work`, and a caller that consumed the cluster + // has no handle left to close it, so each peer gets its close here. + dialers.close().await; + result } } @@ -915,20 +1081,27 @@ struct Work { /// origin. Loop prevention and route preference come from the hop list carried /// on each announced route (see [`moq_net::origin::Route`]). /// -/// Construct with [`Cluster::new`], then attach a QUIC client and (optionally) +/// Construct with [`Cluster::new`], then attach a dial template and (optionally) /// a [`stats::Registry`](moq_net::stats::Registry) with the `with_*` builder /// methods. Those builders do not rebuild the origin. A cluster without a -/// client can serve local sessions but cannot dial remote peers. +/// dial template can serve local sessions but cannot dial remote peers. #[derive(Clone)] pub struct Cluster { session_limits: moq_net::session::Limits, peer_limits: moq_net::session::Limits, + peer_quic: moq_tokio::quic::Limits, config: Config, - client: Option, - /// Dial template for LAN peers that pin a fingerprint; static and API - /// dials keep [`Self::client`]. + /// Dial template every peer dial is built from; see [`Self::peer_client`]. connect: Option, quic: Option, + #[cfg(feature = "iroh")] + iroh: Option, + /// The client static and API peers share, on the template's bind address, so + /// a fixed `[connect]` bind port stays the cluster's egress port. Built by + /// [`Self::start`] and joins [`Self::dialers`] once [`Started::run`] begins. + dial: Option, + /// Every dial client, shared by all clones, for [`Self::close`]. + dialers: Dialers, /// Bound listen port and optional generated-certificate fingerprint. advertise: Option, /// Set by [`Self::start`] when LAN discovery is on, so inbound @@ -997,6 +1170,11 @@ 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) + } + /// The origin ID used by this relay on the wire. pub fn id(&self) -> u64 { self.origin.hop().id() @@ -1005,7 +1183,7 @@ impl Cluster { /// Creates a cluster with one origin, using [`Options`] for identity /// and cache. /// - /// Use [`with_client`](Self::with_client) to enable dialing remote peers + /// Use [`with_connect`](Self::with_connect) to enable dialing remote peers /// (required when `config.connect` is non-empty), and /// [`with_stats`](Self::with_stats) to enable metrics publishing. Those /// builders do not rebuild the origin. @@ -1016,6 +1194,7 @@ impl Cluster { /// /// Errors if `config.id` is set but invalid: it must be non-zero and below /// 2^62 (the wire varint limit). An unset id picks a fresh random origin. + /// Errors too if the `peer.quic` receive window does not fit a QUIC varint. pub fn new(options: Options) -> anyhow::Result { let Options { config, @@ -1055,10 +1234,14 @@ impl Cluster { Ok(Cluster { session_limits: session.resolve(), peer_limits: peer.resolve(), + peer_quic: peer.quic.resolve()?, config, - client: None, connect: None, quic: None, + dial: None, + dialers: Dialers::new(), + #[cfg(feature = "iroh")] + iroh: None, advertise: None, #[cfg(feature = "cluster-lan")] lan_auth: Arc::new(std::sync::OnceLock::new()), @@ -1175,19 +1358,11 @@ impl Cluster { }) } - /// Attach a QUIC client used to dial cluster peers. + /// Attach the dial template every peer dial is built from, with the + /// [`Options::peer`] QUIC limits. Static and API peers share one socket on its + /// bind address; each LAN dial takes an ephemeral port. /// - /// Required when `config.connect` is non-empty; [`start`](Self::start) returns - /// an error otherwise. - pub fn with_client(mut self, client: moq_tokio::Client) -> Self { - self.client = Some(client); - self - } - - /// Attach the dial template used to build a per-peer client for LAN mesh - /// dials (fingerprint pinning, request-path versions, ephemeral bind). - /// - /// Required when `--cluster-lan` is on; [`start`](Self::start) returns an + /// Required when any peer is configured; [`start`](Self::start) returns an /// error otherwise. pub fn with_connect(mut self, connect: moq_tokio::connect::Config, quic: moq_tokio::quic::Config) -> Self { self.connect = Some(connect); @@ -1195,6 +1370,13 @@ impl Cluster { self } + /// Dial `iroh://` peers through this endpoint. + #[cfg(feature = "iroh")] + pub fn with_iroh(mut self, iroh: moq_tokio::iroh::Endpoint) -> Self { + self.iroh = Some(iroh); + self + } + /// Advertise this bound listener on the LAN. /// /// Required when `--cluster-lan` is on; [`start`](Self::start) returns an @@ -1207,9 +1389,9 @@ impl Cluster { /// Attach the client TLS config used for `--cluster-connect-api` peer-list /// fetches. Only its client identity is shared; the API has separate server - /// trust and verifies the hostname in its URL. Required for an HTTP(S) API; pass the same config - /// used to build the QUIC [`with_client`](Self::with_client) so the endpoint - /// sees this relay's cluster certificate. + /// trust and verifies the hostname in its URL. Required for an HTTP(S) API; pass the TLS config of + /// the [`with_connect`](Self::with_connect) template so the endpoint sees this + /// relay's cluster certificate. pub fn with_client_tls(mut self, tls: rustls::ClientConfig) -> Self { self.client_tls = Some(Arc::new(tls)); self @@ -1358,6 +1540,16 @@ impl Cluster { Ok(()) } + /// Close every peer link this cluster dialed, on any transport, and wait until + /// each peer has been sent its close. Call it before the runtime stops: a + /// dropped link only queues its close, so the peer would keep the link and its + /// routes until its idle timeout. Dialing stops for good, on every clone, and + /// every caller waits for the same shutdown. [`Relay::run`](crate::Relay::run) + /// calls it on the way out. + pub async fn close(&self) { + self.dialers.close().await; + } + /// Validate the cluster config and bind anything that can fail, returning a /// [`Started`] to run. /// @@ -1367,9 +1559,9 @@ impl Cluster { /// would release systemd's dependent units on a relay that is about to exit. /// Call this before signalling readiness, and hand the result to `run`. /// - /// Bails when peers are configured to dial but no client was attached via - /// [`with_client`](Self::with_client). - pub async fn start(self) -> anyhow::Result { + /// Bails when peers are configured to dial but no usable dial template was + /// attached via [`with_connect`](Self::with_connect). + pub async fn start(mut self) -> anyhow::Result { let node = self.config.node.clone(); let lan = self.lan(); #[cfg(feature = "cluster-lan")] @@ -1406,11 +1598,6 @@ impl Cluster { }); } - anyhow::ensure!( - self.client.is_some(), - "cluster peers configured but no QUIC client attached (call Cluster::with_client)" - ); - // Build here so malformed API roots fail startup, before background work. let http = if self.config.connect_api.as_deref().is_some_and(connect_api_is_http) { let tls = self @@ -1445,15 +1632,29 @@ impl Cluster { .expect("--cluster-lan needs Cluster::with_advertise"); let secret = self.config.lan.secret.as_deref(); let app = self.config.lan.app.clone().unwrap_or_default(); - let discovery = lan_discovery(advertise, node.as_deref(), secret, app).await?; - let _ = self.lan_auth.set(LanAuth { - credential: discovery.credential().to_string(), - }); - Some(discovery) + Some(lan_discovery(advertise, node.as_deref(), secret, app).await?) } false => None, }; + // Built here, so a bad template fails startup rather than every dial, and + // last, so a start that fails earlier never binds a fixed `[connect]` port. + // Static and API peers share this client; LAN dials each build their own, + // so a LAN-only node builds one only to check the template. + let statics = !self.config.connect.is_empty() || self.config.connect_api.is_some(); + let client = self + .peer_client(!statics, None) + .context("cluster peers configured but the dial template is unusable")?; + self.dial = statics.then_some(client); + + // After everything fallible, so a failed start leaves no credential behind. + #[cfg(feature = "cluster-lan")] + if let Some(discovery) = &discovery { + let _ = self.lan_auth.set(LanAuth { + credential: discovery.credential().to_string(), + }); + } + Ok(Started { work: Some(Work { http, @@ -1720,9 +1921,14 @@ impl Cluster { async fn supervise_remote(self, target: DialTarget, token: String) { let log_peer = target.key.clone(); - if let Err(err) = self.run_remote(&target, token).await { - tracing::warn!(%err, peer = %log_peer, "cluster peer connection ended"); - } + let dialers = self.dialers.clone(); + dialers + .supervise(async move { + if let Err(err) = self.run_remote(&target, token).await { + tracing::warn!(%err, peer = %log_peer, "cluster peer connection ended"); + } + }) + .await; } #[tracing::instrument("remote", skip_all, err, fields(remote = %target.key))] @@ -1792,12 +1998,19 @@ impl Cluster { let redacted = moq_tokio::RedactedUrl::new(first); tracing::info!(url = %redacted, "dialing cluster peer"); - let client = if target.lan { - self.lan_client(target.fingerprint.as_deref())? - } else { - self.client - .clone() - .context("internal: cluster peer dial without an attached QUIC client")? + // A LAN client is this dial's alone, so it closes when the dial ends. + let (client, _held) = match target.lan { + true => { + let client = self.peer_client(true, target.fingerprint.as_deref())?; + let held = self.dialers.hold(&client)?; + (client, Some(held)) + } + false => ( + self.dial + .clone() + .context("internal: cluster peer dial before Cluster::start")?, + None, + ), }; // Cluster dials use their configured stats tier. Cluster peers carry no auth @@ -1845,31 +2058,44 @@ impl Cluster { } } - /// A client for one LAN dial: request-path versions, ephemeral bind, and - /// the advertised fingerprint pinned on a clean TLS config so the relay's - /// CA roots cannot combine with it. - fn lan_client(&self, fingerprint: Option<&str>) -> anyhow::Result { + /// A peer client from the dial template, with the [`Options::peer`] QUIC + /// limits, since every dial is a peer. A LAN client is built per dial, so it + /// takes an ephemeral port beside the shared static client, keeps only + /// request-path versions, and pins the advertised fingerprint on a clean TLS + /// config, so the relay's CA roots cannot combine with it. + fn peer_client(&self, lan: bool, fingerprint: Option<&str>) -> anyhow::Result { let mut connect = self .connect .clone() - .context("internal: LAN dial without Cluster::with_connect")?; - connect.backoff.timeout = Some(std::time::Duration::ZERO); - connect.once = Some(false); - let mut bind = connect.resolve().bind; - bind.set_port(0); - connect.bind = Some(bind); - connect.version = connect - .versions() - .iter() - .filter(|version| Self::carries_request_path(version)) - .copied() - .collect(); + .context("cluster dial without a dial template (call Cluster::with_connect)")?; + if lan { + let mut bind = connect.resolve().bind; + bind.set_port(0); + connect.bind = Some(bind); + connect.backoff.timeout = Some(std::time::Duration::ZERO); + connect.once = Some(false); + connect.version = connect + .versions() + .iter() + .filter(|version| Self::carries_request_path(version)) + .copied() + .collect(); + } if let Some(fingerprint) = fingerprint { connect.tls = moq_tokio::tls::Connect::default(); connect.tls.fingerprint = vec![fingerprint.to_string()]; } - let quic = self.quic.clone().unwrap_or_default(); - Ok(connect.init(quic)?) + 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()); + let client = connect.init(quic)?; + #[cfg(feature = "iroh")] + let client = match &self.iroh { + Some(iroh) => client.with_iroh(iroh.clone()), + None => client, + }; + Ok(client) } /// Start a reconnecting LAN session from discovered details. Tests wire this @@ -1878,7 +2104,7 @@ impl Cluster { fn dial_lan_target(&self, target: &DialTarget) -> anyhow::Result { let addrs = moq_tokio::Addrs::collect(target.addrs()).context("peer advertised no reachable address")?; let mut client = self - .lan_client(target.fingerprint.as_deref())? + .peer_client(true, target.fingerprint.as_deref())? .with_origin(self.origin.clone().peer()); if let Some(cost) = target.cost { client = client.with_cost(cost); @@ -3356,6 +3582,296 @@ mod tests { ); } + /// A start that fails binds no dial socket, so a corrected start on the same + /// fixed `[connect]` port succeeds. One that never runs registers nothing, so + /// dropping it releases its client rather than leaving it to `close`. + #[tokio::test] + async fn a_failed_start_leaves_the_dial_bind_free() { + let _ = moq_tokio::crypto::install_default(); + let dir = tempfile::tempdir().expect("tempdir"); + let token = dir.path().join("token"); + let config = Config { + connect: vec![Peer::new("moql://127.0.0.1:9/")], + token: Some(token.clone()), + ..Default::default() + }; + let mut connect = moq_tokio::connect::Config::default(); + // Borrow a free port from the OS, then hand it to the dial template. + connect.bind = Some( + std::net::UdpSocket::bind("127.0.0.1:0") + .expect("probe socket") + .local_addr() + .expect("probe addr"), + ); + let cluster = new_cluster(config) + .expect("cluster") + .with_connect(connect, Default::default()); + + let err = cluster.clone().start().await.expect_err("the token file is missing"); + assert!(format!("{err:#}").contains("cluster token"), "{err:#}"); + std::fs::write(&token, "secret").expect("write token"); + let _started = cluster.clone().start().await.expect("a corrected start binds the port"); + let registered = cluster.dialers.lock().open.as_ref().expect("open").clients.len(); + assert_eq!(registered, 0, "a start registers its client only once it runs"); + } + + /// A QUIC endpoint that accepts the ALPNs of `connect` and speaks no MoQ, so a + /// test sees the transport close of the dial itself. `idle` outlasts any wait. + #[cfg(feature = "quic")] + fn bare_quic_server( + connect: &moq_tokio::connect::Config, + idle: std::time::Duration, + ) -> moq_tokio::quic::driver::Endpoint { + use moq_tokio::quic::driver; + let key = rcgen::KeyPair::generate().expect("keypair"); + let cert = rcgen::CertificateParams::new(vec!["localhost".to_string()]) + .expect("cert params") + .self_signed(&key) + .expect("self-signed cert"); + let mut tls = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert( + vec![cert.der().clone()], + rustls::pki_types::PrivateKeyDer::Pkcs8(key.serialize_der().into()), + ) + .expect("server tls"); + tls.alpn_protocols = connect + .versions() + .alpns() + .iter() + .map(|alpn| alpn.as_bytes().to_vec()) + .collect(); + let crypto = driver::crypto::rustls::QuicServerConfig::try_from(tls).expect("quic crypto"); + let mut transport = driver::TransportConfig::default(); + transport.max_idle_timeout(Some(idle.try_into().expect("idle timeout"))); + let mut server_config = driver::ServerConfig::with_crypto(Arc::new(crypto)); + server_config.transport_config(Arc::new(transport)); + driver::Endpoint::server(server_config, "127.0.0.1:0".parse().unwrap()).expect("server") + } + + /// Cancelling `run` aborts the dials a peer list started, which run outside + /// the tasks of `run` itself, so none keeps redialing with nothing left to close it. + #[cfg(feature = "quic")] + #[tokio::test] + async fn cancelling_run_stops_its_dynamic_dials() { + const TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + const IDLE: std::time::Duration = std::time::Duration::from_secs(300); + let _ = moq_tokio::crypto::install_default(); + + let mut connect = moq_tokio::connect::Config::default(); + connect.tls.insecure = Some(true); + let server = bare_quic_server(&connect, IDLE); + let port = server.local_addr().expect("server addr").port(); + let dir = tempfile::tempdir().expect("tempdir"); + let peers = dir.path().join("peers.json"); + std::fs::write(&peers, format!(r#"["moql://127.0.0.1:{port}/"]"#)).expect("write peers"); + + let config = Config { + connect_api: Some(peers.display().to_string()), + ..Default::default() + }; + let mut quic = moq_tokio::quic::Config::default(); + quic.idle_timeout = Some(IDLE); + let cluster = new_cluster(config).expect("cluster").with_connect(connect, quic); + let started = cluster.clone().start().await.expect("start"); + let running = tokio::spawn(started.run()); + + let _conn = tokio::time::timeout(TIMEOUT, server.accept()) + .await + .expect("dial timeout") + .expect("incoming") + .await + .expect("handshake"); + running.abort(); + let _ = running.await; + // Each dial holds a stop receiver until its task ends. + tokio::time::timeout(TIMEOUT, cluster.dialers.stop.closed()) + .await + .expect("a peer-list dial outlived its cancelled run"); + } + + /// A critical failure ends `run`, which closes the dials on the way out: the + /// caller consumed the cluster, so nothing else could. The dialing runtime stops + /// right after, as a process exit would, and the peer's idle timeout is far past + /// the wait, so only the close explains the link ending. + #[cfg(feature = "quic")] + #[tokio::test] + async fn a_critical_failure_closes_the_dials() { + use moq_tokio::quic::driver; + const TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + const IDLE: std::time::Duration = std::time::Duration::from_secs(300); + let _ = moq_tokio::crypto::install_default(); + + let mut connect = moq_tokio::connect::Config::default(); + connect.tls.insecure = Some(true); + let server = bare_quic_server(&connect, IDLE); + let port = server.local_addr().expect("server addr").port(); + + let config = Config { + connect: vec![Peer::new(format!("moql://127.0.0.1:{port}/"))], + ..Default::default() + }; + let mut quic = moq_tokio::quic::Config::default(); + quic.idle_timeout = Some(IDLE); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .expect("dialer runtime"); + let (cluster, running) = runtime + .spawn(async move { + let cluster = new_cluster(config).expect("cluster").with_connect(connect, quic); + let started = cluster.clone().start().await.expect("start"); + (cluster, tokio::spawn(started.run())) + }) + .await + .expect("dialer task"); + + let conn = tokio::time::timeout(TIMEOUT, server.accept()) + .await + .expect("dial timeout") + .expect("incoming") + .await + .expect("handshake"); + cluster + .critical + .spawn("injected", async { panic!("injected critical failure") }); + let result = tokio::time::timeout(TIMEOUT, running) + .await + .expect("run returns on a critical failure") + .expect("run task"); + assert!(result.is_err(), "a critical failure fails run"); + runtime.shutdown_background(); + + let err = tokio::time::timeout(TIMEOUT, conn.closed()) + .await + .expect("the peer heard no close before its idle timeout"); + match err { + driver::ConnectionError::ApplicationClosed(close) => assert_eq!(&close.reason[..], b"client shutdown"), + other => panic!("the link ended with {other:?}, not the shutdown close"), + } + } + + /// A second close awaits the shutdown the first started, and still finishes it + /// when the first caller gives up, so no caller returns while links are open. + #[tokio::test] + async fn concurrent_closes_share_one_shutdown() { + let dialers = Dialers::new(); + // Stands in for a dropped guard's background close, which holds the shutdown open. + let background = dialers.lock().open.as_ref().expect("open").closing.clone(); + let first = tokio::spawn({ + let dialers = dialers.clone(); + async move { dialers.close().await } + }); + while dialers.lock().open.is_some() { + tokio::task::yield_now().await; + } + + let mut second = std::pin::pin!(dialers.close()); + assert!( + futures::poll!(second.as_mut()).is_pending(), + "a second close returned before the shutdown finished" + ); + first.abort(); + let _ = first.await; + drop(background); + tokio::time::timeout(std::time::Duration::from_secs(5), second) + .await + .expect("the second close finishes the shutdown"); + } + + /// A LAN dial owns its client, so closing the cluster closes that too: the peer + /// drops the dialer's routes at once, though the dialing runtime stops right + /// after, as a process exit would. Both idle timeouts are far past the wait. + #[cfg(feature = "cluster-lan")] + #[tokio::test] + async fn closing_a_cluster_closes_its_lan_dials() { + const TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + const IDLE: std::time::Duration = std::time::Duration::from_secs(300); + let _ = moq_tokio::crypto::install_default(); + let mut quic = moq_tokio::quic::Config::default(); + quic.idle_timeout = Some(IDLE); + + let node = new_cluster(Config::default()).expect("node cluster"); + let mut listen = moq_tokio::listen::Config::default(); + listen.bind = Some("127.0.0.1:0".parse().unwrap()); + listen.tls.generate = vec!["moq-cluster-lan".to_string()]; + let server = listen.init(quic.clone()).expect("bind"); + let port = server.local_addr().expect("local addr").port(); + let fp = server + .certificates() + .fingerprints() + .into_iter() + .next() + .expect("generated fingerprint"); + let listener = server.listen().await.expect("listen"); + node.set_lan_credential("listener-proof"); + let accept = node.clone(); + tokio::spawn(async move { + let mut listener = listener; + while let Some(request) = listener.accept().await { + let conn = crate::Connection::new(request, accept.clone(), crate::auth::Auth::refuse("test")); + tokio::spawn(conn.run()); + } + }); + + let url: Url = format!("moqt://127.0.0.1:{port}{CLUSTER_PATH}/listener-proof") + .parse() + .expect("url"); + let target = DialTarget { + key: format!("moqt://127.0.0.1:{port}/"), + url: url.clone(), + urls: vec![url], + cost: None, + fingerprint: Some(fp), + lan: true, + upstream: false, + }; + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .expect("dialer runtime"); + let (dialer, broadcast) = runtime + .spawn(async move { + let dialer = new_cluster(Config::default()) + .expect("dialer cluster") + .with_connect(Default::default(), quic); + let broadcast = dialer.origin.create_broadcast("from-dialer").expect("create"); + broadcast.announce(Default::default()).expect("announce"); + tokio::spawn(dialer.clone().supervise_remote(target, String::new())); + (dialer, broadcast) + }) + .await + .expect("dialer task"); + + let mut announced = node.origin.consume().announced(); + loop { + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announced)) + .await + .expect("timed out waiting for from-dialer") + .expect("origin closed"); + if update.prefix.as_str() == "from-dialer" && active { + break; + } + } + + runtime.spawn(async move { dialer.close().await }).await.expect("close"); + runtime.shutdown_background(); + // Held throughout: dropping it would unannounce over a live link instead. + let _broadcast = broadcast; + + loop { + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announced)) + .await + .expect("the peer kept the dialer's route past the close") + .expect("origin closed"); + if update.prefix.as_str() == "from-dialer" && !active { + break; + } + } + } + /// A grant naming a cluster peer marks the session's routes as a peer's, so the /// relay's local view leaves them out; any other grant ingests here. #[tokio::test] diff --git a/rs/moq-relay/src/config.rs b/rs/moq-relay/src/config.rs index e48a0669959..2944ea3b78f 100644 --- a/rs/moq-relay/src/config.rs +++ b/rs/moq-relay/src/config.rs @@ -1,3 +1,4 @@ +use anyhow::Context; use serde::{Deserialize, Serialize}; use crate::{auth, cache, cluster, internal, stats, web}; @@ -361,6 +362,10 @@ impl Config { pub(crate) fn resolve(&mut self) -> anyhow::Result<()> { let deprecated = self.deprecated(); anyhow::ensure!(deprecated.is_empty(), "{deprecated}"); + // Here rather than at an endpoint, since io_uring workers and lazily built + // dial clients would otherwise check `[quic]` late or never. `[peer.quic]` + // is checked by `Cluster::new`, which embedders call directly. + self.quic.validate().context("invalid [quic]")?; self.quic .max_streams @@ -441,6 +446,62 @@ mod tests { assert_eq!(config.peer.resolve().fetches, 2000); } + /// `[peer.quic]` mirrors `[quic]`'s keys without touching `[quic]` itself. + #[test] + fn peer_quic_table_merges_with_flags() { + let _env = EnvGuard::clear(&[ + "MOQ_QUIC_MAX_STREAMS", + "MOQ_PEER_QUIC_MAX_STREAMS", + "MOQ_PEER_QUIC_RECEIVE_WINDOW", + "MOQ_PEER_QUIC_SEND_WINDOW", + ]); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("peer-quic.toml"); + std::fs::write( + &path, + "[quic]\nmax_streams = 4\n[peer.quic]\nmax_streams = 64\nsend_window = 1000\n", + ) + .unwrap(); + let config = Config::parse_and_merge([ + std::ffi::OsString::from("moq-relay"), + path.into(), + "--peer-quic-receive-window=2000".into(), + ]) + .unwrap(); + assert_eq!(config.quic.max_streams, Some(4)); + assert_eq!(config.peer.quic.max_streams.get(), 64); + assert_eq!(config.peer.quic.receive_window.get(), 2000); + assert_eq!(config.peer.quic.send_window.get(), 1000); + + assert!(toml::from_str::("[peer]\nmax_streams = 64\n").is_err()); + } + + /// A zero peer stream limit or window would wedge the connection, so it never parses. + #[test] + fn zero_peer_quic_limits_are_refused() { + let _env = EnvGuard::clear(&[ + "MOQ_PEER_QUIC_MAX_STREAMS", + "MOQ_PEER_QUIC_RECEIVE_WINDOW", + "MOQ_PEER_QUIC_SEND_WINDOW", + ]); + for key in ["max_streams", "receive_window", "send_window"] { + assert!( + toml::from_str::(&format!("[peer.quic]\n{key} = 0\n")).is_err(), + "{key}" + ); + } + for flag in [ + "--peer-quic-max-streams=0", + "--peer-quic-receive-window=0", + "--peer-quic-send-window=0", + ] { + assert!( + Config::parse_and_merge([std::ffi::OsString::from("moq-relay"), flag.into()]).is_err(), + "{flag} must be refused" + ); + } + } + #[cfg(not(feature = "iroh"))] #[test] fn iroh_environment_requires_the_feature() { diff --git a/rs/moq-relay/src/connection.rs b/rs/moq-relay/src/connection.rs index c7681688842..cced96da4b0 100644 --- a/rs/moq-relay/src/connection.rs +++ b/rs/moq-relay/src/connection.rs @@ -124,6 +124,13 @@ impl Connection { stats, } = admitted; + // A peer's QUIC limits widen before SETUP OK, so its first requests see them. + if let Some(limits) = self.cluster.quic_limits(&lease) + && let Err(err) = self.request.set_quic_limits(limits) + { + tracing::debug!(%err, "cluster peer keeps its transport's limits"); + } + // Accept the connection. // NOTE: subscribe and publish seem backwards because of how relays work. // We publish the tracks the client is allowed to subscribe to. diff --git a/rs/moq-relay/src/limits.rs b/rs/moq-relay/src/limits.rs index 2227ae61c71..09aab2539ec 100644 --- a/rs/moq-relay/src/limits.rs +++ b/rs/moq-relay/src/limits.rs @@ -1,5 +1,6 @@ -//! Request caps selected after the relay authorizes a session. +//! Request caps and QUIC limits selected after the relay authorizes a session. +use anyhow::Context; use serde::{Deserialize, Serialize}; /// Request caps for ordinary client sessions. @@ -49,7 +50,7 @@ impl Session { } } -/// Request caps for authenticated cluster peer sessions. +/// Request caps and QUIC limits for authenticated cluster peer sessions. #[derive(usage::Args, Clone, Debug, Deserialize, Serialize)] #[serde(default, deny_unknown_fields)] #[non_exhaustive] @@ -78,6 +79,9 @@ pub struct Peer { setting = "peer.fetches" )] pub fetches: usize, + /// QUIC limits a peer session widens to (`[peer.quic]`). + #[usage(flatten)] + pub quic: Quic, } impl Default for Peer { @@ -86,6 +90,7 @@ impl Default for Peer { announces: 100000, subscriptions: 10000, fetches: 10000, + quic: Quic::default(), } } } @@ -96,6 +101,64 @@ impl Peer { } } +/// QUIC limits for peer sessions and the links the relay dials, keyed like `[quic]`. +#[derive(usage::Args, Clone, Debug, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +#[non_exhaustive] +pub struct Quic { + /// Maximum concurrent QUIC streams the peer may open, per direction. + #[usage( + long = "peer-quic-max-streams", + env = "MOQ_PEER_QUIC_MAX_STREAMS", + default = "100000", + setting = "peer.quic.max_streams" + )] + pub max_streams: std::num::NonZeroU64, + /// Connection-wide QUIC receive window, in bytes. + #[usage( + long = "peer-quic-receive-window", + env = "MOQ_PEER_QUIC_RECEIVE_WINDOW", + default = "671088640", + setting = "peer.quic.receive_window" + )] + pub receive_window: std::num::NonZeroU64, + /// Cap on unacknowledged outgoing QUIC data, in bytes. + #[usage( + long = "peer-quic-send-window", + env = "MOQ_PEER_QUIC_SEND_WINDOW", + default = "671088640", + setting = "peer.quic.send_window" + )] + pub send_window: std::num::NonZeroU64, +} + +impl Default for Quic { + fn default() -> Self { + // An order of magnitude above the `[quic]` defaults, which are sized for + // one viewer rather than a link carrying a whole cluster's traffic. + let window = std::num::NonZeroU64::new(10 * moq_tokio::quic::DEFAULT_RECEIVE_WINDOW).expect("non-zero"); + Self { + max_streams: std::num::NonZeroU64::new(10 * moq_tokio::quic::DEFAULT_MAX_STREAMS).expect("non-zero"), + receive_window: window, + send_window: window, + } + } +} + +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) + } +} + fn resolve(announces: usize, subscriptions: usize, fetches: usize) -> moq_net::session::Limits { let mut limits = moq_net::session::Limits::default(); limits.announces = announces; diff --git a/rs/moq-relay/src/relay.rs b/rs/moq-relay/src/relay.rs index 71f5d571e6b..e9f30067d1d 100644 --- a/rs/moq-relay/src/relay.rs +++ b/rs/moq-relay/src/relay.rs @@ -1,7 +1,7 @@ //! The assembled relay: every startup step, in one place, in the right order. //! -//! [`Relay::load`] turns a [`Config`] into the running pieces (server, client, -//! auth, cluster, stats, internal + web listeners), and [`Relay::run`] drives +//! [`Relay::load`] turns a [`Config`] into the running pieces (server, auth, +//! cluster, stats, internal + web listeners), and [`Relay::run`] drives //! them. `main.rs` is a thin wrapper over the two. //! //! # Why this is a type and not just `main` @@ -50,7 +50,7 @@ impl Ready { /// shutdown join. /// /// The accessors borrow, and [`Self::run`] consumes the relay, so clone the -/// application handles ([`Self::cluster`], [`Self::auth`], [`Self::client`], +/// application handles ([`Self::cluster`], [`Self::auth`], /// [`Self::stats`], [`Self::shutdown`], [`Self::shutdown_trigger`]) first, /// then mount extra routes with [`Self::with_web`] / [`Self::with_internal`]. /// An application that decides admissions itself leaves `[auth]` empty and @@ -72,7 +72,6 @@ pub struct Relay { ready: tokio::sync::watch::Sender, config: Config, server: moq_tokio::Listener, - client: moq_tokio::Client, auth: auth::Auth, /// The sessions the embedder decides, until it takes them. `None` when the /// config named a source; `run` refuses to start while this is still held. @@ -245,7 +244,6 @@ impl Relay { true => server_config.init_streams()?, false => server_config.init()?, }; - let client = config.connect.clone().init(config.quic.clone())?; // `None` for a stream-only server (no QUIC); any other error is real. #[cfg(all(target_os = "linux", feature = "_uring"))] @@ -262,12 +260,6 @@ impl Relay { }, }; - #[cfg(feature = "iroh")] - let client = match iroh { - Some(iroh) => client.with_iroh(iroh), - None => client, - }; - let cache = config.cache.init()?; // Whichever worker group owns QUIC holds the certificates; the shared // server then has none of its own. @@ -289,10 +281,14 @@ impl Relay { cluster_options.session = config.session; cluster_options.peer = config.peer; let cluster = cluster::Cluster::new(cluster_options)? - .with_client(client.clone()) .with_client_tls(config.connect.tls.build()?) .with_connect(config.connect.clone(), config.quic.clone()) .with_advertise(advertise); + #[cfg(feature = "iroh")] + let cluster = match iroh { + Some(iroh) => cluster.with_iroh(iroh), + None => cluster, + }; let stats = config.stats.build(cluster.origin.clone()); // The cluster takes over keeping the publish task alive, so an embedder that // drops the producer keeps publishing for as long as it serves. @@ -348,7 +344,6 @@ impl Relay { ready, config: resolved_config, server, - client, auth, admissions, cluster, @@ -402,12 +397,6 @@ impl Relay { self.server.tcp_local_addr() } - /// The client used to dial cluster peers. Already handed to [`Self::cluster`]; - /// clone it for your own outbound dials so they share the connection config. - pub fn client(&self) -> &moq_tokio::Client { - &self.client - } - /// Where every session's grant comes from: the auth server, the static public /// grant, or the embedder answering [`Self::admissions`]. Clone it to admit /// your own listeners' sessions the same way. @@ -697,6 +686,9 @@ impl Relay { else => Ok(()), }; + // Before the runtime goes, so each dialed peer drops our routes at once. + cluster.close().await; + // Dropping the listener would leave its QUIC socket open until the // endpoint's closing connections finish in the background, past `run`. server.close().await; diff --git a/rs/moq-relay/src/settings.rs b/rs/moq-relay/src/settings.rs index c13d5d0f207..3c2efa1875d 100644 --- a/rs/moq-relay/src/settings.rs +++ b/rs/moq-relay/src/settings.rs @@ -238,6 +238,19 @@ struct Peer { subscriptions: Option, #[usage(env = "MOQ_PEER_FETCHES", cli("--peer-fetches"))] fetches: Option, + #[usage(flatten)] + quic: PeerQuic, +} + +#[derive(usage::Config)] +#[usage(prefix = "peer.quic")] +struct PeerQuic { + #[usage(env = "MOQ_PEER_QUIC_MAX_STREAMS", cli("--peer-quic-max-streams"))] + max_streams: Option, + #[usage(env = "MOQ_PEER_QUIC_RECEIVE_WINDOW", cli("--peer-quic-receive-window"))] + receive_window: Option, + #[usage(env = "MOQ_PEER_QUIC_SEND_WINDOW", cli("--peer-quic-send-window"))] + send_window: Option, } #[derive(usage::Config)] diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 5f8d6b4a7e1..f938f9658bc 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -681,6 +681,7 @@ async fn serve_connection( }; deadline.close = Some(transport.clone()); + let conn = transport.clone(); let server = moq_net::Server::new().with_versions(serve.versions.clone()); let binding = match url { Some(_) => moq_net::transport::Binding::Uri, @@ -808,6 +809,14 @@ async fn serve_connection( stats, } = admitted; + // A peer's QUIC limits widen before SETUP OK, so its first requests see them. + if let Some(limits) = serve.cluster.quic_limits(&lease) { + conn.set_limits(moq_uring::quic::Limits { + max_streams: limits.max_streams, + receive_window: limits.receive_window, + send_window: limits.send_window, + })?; + } let mut request = request .with_stats(stats.clone()) .with_limits(serve.cluster.limits(&lease)); diff --git a/rs/moq-relay/tests/integration/cluster_close.rs b/rs/moq-relay/tests/integration/cluster_close.rs new file mode 100644 index 00000000000..798b8784ad6 --- /dev/null +++ b/rs/moq-relay/tests/integration/cluster_close.rs @@ -0,0 +1,163 @@ +//! A cluster sends each dialed peer its close before the runtime stops, so the +//! peer drops the link and its routes at once instead of at its idle timeout. +//! +//! Each dialer runs on its own runtime, shut down right after, as a process exit +//! would: a dropped link only queues its close, which nothing sends then. +#![cfg(feature = "quic")] + +use std::time::Duration; + +use moq_tokio::quic::driver; + +use crate::peer_limits::{BARE_IDLE, bare_server, config}; + +/// Far short of [`BARE_IDLE`] and the dial's idle timeout, so only a close explains +/// the link ending in time. +const TIMEOUT: Duration = Duration::from_secs(10); + +fn dialer_runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .expect("dialer runtime") +} + +fn dial_template() -> (moq_tokio::connect::Config, moq_tokio::quic::Config) { + let mut connect = moq_tokio::connect::Config::default(); + connect.tls.insecure = Some(true); + let mut quic = moq_tokio::quic::Config::default(); + quic.idle_timeout = Some(BARE_IDLE); + (connect, quic) +} + +fn peer_url(server: &driver::Endpoint) -> String { + let port = server.local_addr().expect("server addr").port(); + format!("moql://127.0.0.1:{port}/") +} + +async fn accept(server: &driver::Endpoint) -> driver::Connection { + tokio::time::timeout(TIMEOUT, server.accept()) + .await + .expect("dial timeout") + .expect("incoming") + .await + .expect("handshake") +} + +/// The link ends with the dialer's own shutdown close, not a timeout or a SETUP error. +async fn assert_closed_by_shutdown(conn: driver::Connection) { + let err = tokio::time::timeout(TIMEOUT, conn.closed()) + .await + .expect("the peer heard no close before its idle timeout"); + match err { + driver::ConnectionError::ApplicationClosed(close) => assert_eq!(&close.reason[..], b"client shutdown"), + other => panic!("the link ended with {other:?}, not the shutdown close"), + } +} + +#[tokio::test] +async fn closing_a_cluster_closes_its_static_dials() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let (connect, quic) = dial_template(); + let server = bare_server(&connect); + + let mut options = moq_relay::cluster::Options::new(Default::default()); + options.config.connect = vec![moq_relay::cluster::Peer::new(peer_url(&server))]; + let runtime = dialer_runtime(); + let cluster = runtime + .spawn(async move { + let cluster = moq_relay::cluster::Cluster::new(options) + .expect("cluster init") + .with_connect(connect, quic); + let started = cluster.clone().start().await.expect("cluster start"); + tokio::spawn(started.run()); + cluster + }) + .await + .expect("cluster task"); + + let conn = accept(&server).await; + runtime + .spawn(async move { cluster.close().await }) + .await + .expect("close"); + runtime.shutdown_background(); + + assert_closed_by_shutdown(conn).await; +} + +/// `Client::close` reaches only QUIC, so a WebSocket link ends only because the +/// close stops its dial task. The runtime keeps running, as an embedder's would. +#[cfg(feature = "websocket")] +#[tokio::test] +async fn closing_a_cluster_ends_its_websocket_dials() { + use tokio::io::AsyncReadExt; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let port = listener.local_addr().expect("addr").port(); + let mut options = moq_relay::cluster::Options::new(Default::default()); + options.config.connect = vec![moq_relay::cluster::Peer::new(format!("ws://127.0.0.1:{port}/"))]; + let (connect, quic) = dial_template(); + let cluster = moq_relay::cluster::Cluster::new(options) + .expect("cluster init") + .with_connect(connect, quic); + let started = cluster.clone().start().await.expect("cluster start"); + let running = tokio::spawn(started.run()); + + // Never answers the upgrade, so the dial holds the socket until stopped. + let (mut socket, _) = tokio::time::timeout(TIMEOUT, listener.accept()) + .await + .expect("dial timeout") + .expect("accept"); + tokio::time::timeout(TIMEOUT, cluster.close()) + .await + .expect("close waits on a stopped dial"); + + let mut buf = vec![0; 4096]; + loop { + // Brief: the socket is already closed once `close` returns. + let read = tokio::time::timeout(Duration::from_secs(2), socket.read(&mut buf)) + .await + .expect("the dial still holds its socket"); + if read.map_or(true, |n| n == 0) { + break; + } + } + running.abort(); +} + +#[tokio::test] +async fn a_relay_shutdown_closes_its_cluster_dials() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let (connect, quic) = dial_template(); + let server = bare_server(&connect); + + let dir = tempfile::tempdir().expect("tempdir"); + let mut config = config(dir.path()); + config.connect = connect; + config.quic.idle_timeout = quic.idle_timeout; + config.cluster.connect = vec![moq_relay::cluster::Peer::new(peer_url(&server))]; + config.auth.public = vec![moq_auth::Pattern::all()]; + + let runtime = dialer_runtime(); + let (trigger, running) = runtime + .spawn(async move { + let relay = moq_relay::Relay::load(config).await.expect("load relay"); + let trigger = relay.shutdown_trigger().clone(); + (trigger, tokio::spawn(relay.run())) + }) + .await + .expect("relay task"); + + let conn = accept(&server).await; + trigger.start(); + tokio::time::timeout(TIMEOUT, running) + .await + .expect("relay drain") + .expect("relay task") + .expect("relay run"); + runtime.shutdown_background(); + + assert_closed_by_shutdown(conn).await; +} diff --git a/rs/moq-relay/tests/integration/cluster_idle.rs b/rs/moq-relay/tests/integration/cluster_idle.rs index d2bf26cfdb6..f194d664bd5 100644 --- a/rs/moq-relay/tests/integration/cluster_idle.rs +++ b/rs/moq-relay/tests/integration/cluster_idle.rs @@ -95,13 +95,12 @@ async fn silent_peer_detected() { { connect.websocket.enabled = Some(false); } - let client = connect.init(Default::default()).expect("client init"); let mut config = cluster::Config::default(); config.connect = vec![Peer::new(format!("https://127.0.0.1:{port}/"))]; let cluster = cluster::Cluster::new(cluster::Options::new(config)) .expect("cluster init") - .with_client(client); + .with_connect(connect, Default::default()); let mut announced = cluster.origin.consume().announced(); let run = tokio::spawn(cluster.clone().start().await.expect("cluster start").run()); diff --git a/rs/moq-relay/tests/integration/goaway_cluster.rs b/rs/moq-relay/tests/integration/goaway_cluster.rs index f556c56b88f..a5c170280f6 100644 --- a/rs/moq-relay/tests/integration/goaway_cluster.rs +++ b/rs/moq-relay/tests/integration/goaway_cluster.rs @@ -150,13 +150,12 @@ async fn cluster_continues_a_group_split_by_goaway_inner() { let mut client_config = moq_tokio::connect::Config::default(); client_config.tls.insecure = Some(true); client_config.goaway.handover = Some(Duration::from_secs(2)); - let client = client_config.init(Default::default()).expect("client init"); let mut cluster_config = cluster::Config::default(); cluster_config.connect = vec![Peer::new(format!("tcp://127.0.0.1:{port_a}/"))]; let cluster = cluster::Cluster::new(cluster::Options::new(cluster_config)) .expect("cluster init") - .with_client(client); + .with_connect(client_config, Default::default()); let started = cluster.clone().start().await.expect("cluster start"); let cluster_run = tokio::spawn(started.run()); @@ -295,13 +294,12 @@ async fn cluster_migrates_on_upstream_goaway_inner() { client_config.tls.insecure = Some(true); // Short handover so the test observes the old session close quickly. client_config.goaway.handover = Some(Duration::from_secs(2)); - let client = client_config.init(Default::default()).expect("client init"); let mut cluster_config = cluster::Config::default(); cluster_config.connect = vec![Peer::new(format!("tcp://127.0.0.1:{port_a}/"))]; let cluster = cluster::Cluster::new(cluster::Options::new(cluster_config)) .expect("cluster init") - .with_client(client); + .with_connect(client_config, Default::default()); let started = cluster.clone().start().await.expect("cluster start"); let cluster_run = tokio::spawn(started.run()); @@ -422,11 +420,10 @@ async fn spawn_relay_with_upstream( client_config.goaway.handover = Some(Duration::from_secs(2)); // Only lite-07 carries the epoch that lets a failover resume the subscription. client_config.version = vec![LITE_07.parse().unwrap()]; - let client = client_config.init(Default::default()).expect("client init"); let cluster = cluster::Cluster::new(cluster::Options::new(cluster_config)) .expect("cluster init") - .with_client(client); + .with_connect(client_config, Default::default()); let started = cluster.clone().start().await.expect("cluster start"); let handle = tokio::spawn(async move { @@ -771,13 +768,12 @@ async fn cluster_reconnects_on_empty_uri_goaway_inner() { client_config.goaway.handover = Some(Duration::from_secs(2)); // Only lite-07 carries the epoch that lets the redial resume the subscription. client_config.version = vec![LITE_07.parse().unwrap()]; - let client = client_config.init(Default::default()).expect("client init"); let mut cluster_config = cluster::Config::default(); cluster_config.connect = vec![Peer::new(format!("tcp://127.0.0.1:{port}/"))]; let cluster = cluster::Cluster::new(cluster::Options::new(cluster_config)) .expect("cluster init") - .with_client(client); + .with_connect(client_config, Default::default()); let started = cluster.clone().start().await.expect("cluster start"); let cluster_run = tokio::spawn(started.run()); diff --git a/rs/moq-relay/tests/integration/hidden_cluster.rs b/rs/moq-relay/tests/integration/hidden_cluster.rs index 6b7159bf95d..ae19fd9ff59 100644 --- a/rs/moq-relay/tests/integration/hidden_cluster.rs +++ b/rs/moq-relay/tests/integration/hidden_cluster.rs @@ -63,12 +63,11 @@ async fn old_peer_keeps_hidden_paths(version: moq_net::Version) { let mut client_config = moq_tokio::connect::Config::default(); client_config.tls.insecure = Some(true); - let client = client_config.init(Default::default()).expect("client init"); let mut cluster_config = cluster::Config::default(); cluster_config.connect = vec![Peer::new(format!("tcp://127.0.0.1:{port}/"))]; let cluster = cluster::Cluster::new(cluster::Options::new(cluster_config)) .expect("cluster init") - .with_client(client); + .with_connect(client_config, Default::default()); let node = cluster.origin.create_broadcast(".hidden/test").expect("create hidden"); node.announce(Default::default()).expect("announce hidden"); let cluster_run = tokio::spawn(cluster.clone().start().await.expect("cluster start").run()); diff --git a/rs/moq-relay/tests/integration/lan_mesh.rs b/rs/moq-relay/tests/integration/lan_mesh.rs index 0d8dbf5ee87..e39e20cc9e0 100644 --- a/rs/moq-relay/tests/integration/lan_mesh.rs +++ b/rs/moq-relay/tests/integration/lan_mesh.rs @@ -41,7 +41,6 @@ async fn moq_import_cluster_lan_beside_a_relay() { let mut connect = moq_tokio::connect::Config::default(); connect.tls.fingerprint = vec![fingerprint.clone()]; - let client = connect.clone().init(Default::default()).expect("client"); let mut lan = cluster::LanConfig::default(); lan.enabled = true; @@ -51,7 +50,6 @@ async fn moq_import_cluster_lan_beside_a_relay() { let cluster = cluster::Cluster::new(cluster::Options::new(config)) .expect("cluster") - .with_client(client) .with_connect(connect, Default::default()) .with_advertise(cluster::LanAdvertise::new(port).with_fingerprint(fingerprint)); diff --git a/rs/moq-relay/tests/integration/main.rs b/rs/moq-relay/tests/integration/main.rs index 93b3e513909..a21f62cd43c 100644 --- a/rs/moq-relay/tests/integration/main.rs +++ b/rs/moq-relay/tests/integration/main.rs @@ -7,12 +7,14 @@ mod auth_embedder; mod auth_lifetime; +mod cluster_close; mod cluster_idle; mod cluster_unknown; mod embed; mod goaway_cluster; mod hidden_cluster; mod lan_mesh; +mod peer_limits; mod public_rules; mod refusals; mod released_cli; diff --git a/rs/moq-relay/tests/integration/peer_limits.rs b/rs/moq-relay/tests/integration/peer_limits.rs new file mode 100644 index 00000000000..d9ab129570d --- /dev/null +++ b/rs/moq-relay/tests/integration/peer_limits.rs @@ -0,0 +1,326 @@ +//! A session admitted as a cluster peer widens its QUIC limits before SETUP OK, +//! while a viewer keeps the `--quic-*` ones, on the tokio listener and the +//! io_uring workers alike. +#![cfg(feature = "quic")] + +use std::time::Duration; + +use futures::FutureExt; +use moq_relay::{Config, Relay}; +use moq_tokio::quic::driver; + +const TIMEOUT: Duration = Duration::from_secs(10); + +/// The viewer stream limit. Small, so a peer's widened one is unmistakable. +const VIEWER_STREAMS: u64 = 4; +/// How many streams the probe tries to open without waiting for credit. +const PROBE: usize = 16; +/// The [`bare_server`] idle timeout. +pub(crate) const BARE_IDLE: Duration = Duration::from_secs(300); + +/// A self-signed certificate on disk. +fn certificate(dir: &std::path::Path) -> (std::path::PathBuf, std::path::PathBuf) { + let key = rcgen::KeyPair::generate().expect("keypair"); + let params = rcgen::CertificateParams::new(vec!["localhost".to_string()]).expect("cert params"); + let cert = params.self_signed(&key).expect("self-signed cert"); + let cert_path = dir.join("cert.pem"); + let key_path = dir.join("key.pem"); + std::fs::write(&cert_path, cert.pem()).expect("write cert"); + std::fs::write(&key_path, key.serialize_pem()).expect("write key"); + (cert_path, key_path) +} + +pub(crate) fn config(dir: &std::path::Path) -> Config { + let (cert, key) = certificate(dir); + let mut config = Config::default(); + config.listen.bind = Some("127.0.0.1:0".parse().unwrap()); + config.listen.tls.cert = vec![cert]; + config.listen.tls.key = vec![key]; + config.quic.max_streams = Some(VIEWER_STREAMS); + config.peer.quic.max_streams = std::num::NonZeroU64::new(4 * PROBE as u64).unwrap(); + config +} + +/// What a raw QUIC client sees after SETUP: how many unidirectional streams it +/// can open at once, and how many MAX_DATA frames the relay sent. +struct Seen { + streams: usize, + max_data: u64, +} + +/// Dial with an empty moq-lite SETUP, wait for the relay's SETUP, then probe. +/// +/// A peer waits for each stream's credit, since SETUP and the MAX_STREAMS +/// queued before it can travel in separate packets. A viewer takes only the +/// credit it holds once SETUP arrives: the relay widens before that, so any +/// widening would already show. +async fn probe(port: u16, peer: bool) -> Seen { + let mut connect = moq_tokio::connect::Config::default(); + connect.tls.insecure = Some(true); + let mut tls = connect.tls.build().expect("tls config"); + tls.alpn_protocols = vec![b"moq-lite-06".to_vec()]; + let crypto = driver::crypto::rustls::QuicClientConfig::try_from(tls).expect("quic crypto"); + let endpoint = driver::Endpoint::client("127.0.0.1:0".parse().unwrap()).expect("endpoint"); + let conn = endpoint + .connect_with( + driver::ClientConfig::new(std::sync::Arc::new(crypto)), + ([127, 0, 0, 1], port).into(), + "localhost", + ) + .expect("connect") + .await + .expect("handshake"); + + // SETUP type, then a one-byte message holding zero parameters. + let mut setup = conn.open_uni().await.expect("open"); + setup.write_all(&[0x01, 0x01, 0x00]).await.expect("write"); + setup.finish().expect("finish"); + + // The relay widens a peer before it answers SETUP, so its SETUP stream + // arrives behind the new credit. + let _setup = tokio::time::timeout(TIMEOUT, conn.accept_uni()) + .await + .expect("relay SETUP timeout") + .expect("relay SETUP"); + + let mut streams = Vec::new(); + while streams.len() < PROBE { + let open = conn.open_uni(); + let stream = match peer { + true => tokio::time::timeout(TIMEOUT, open) + .await + .expect("peer stream credit timeout"), + false => match open.now_or_never() { + Some(stream) => stream, + None => break, + }, + }; + streams.push(stream.expect("open")); + } + let seen = Seen { + streams: streams.len(), + max_data: conn.stats().frame_rx.max_data, + }; + conn.close(driver::VarInt::from_u32(0), b"done"); + endpoint.wait_idle().await; + seen +} + +/// Admit the first session as a cluster peer and the second as a viewer, and +/// check that only the peer saw wider limits. +async fn peer_widens_and_viewer_does_not(mut relay: Relay, port: u16) { + let mut admissions = relay.admissions().expect("embedded auth"); + let deciding = tokio::spawn(async move { + let mut leases = Vec::new(); + for peer in [true, false] { + let admission = tokio::time::timeout(TIMEOUT, admissions.next()) + .await + .expect("admission timeout") + .expect("admission"); + let all: moq_auth::Patterns = [moq_auth::Pattern::all()].into_iter().collect(); + let mut grant = moq_auth::Grant::new(all.clone(), all); + grant.peer = peer; + let (producer, consumer) = moq_auth::lease::Producer::new(grant); + leases.push(producer); + admission.grant(consumer); + } + leases + }); + let running = tokio::spawn(relay.run()); + + let peer = probe(port, true).await; + assert_eq!(peer.streams, PROBE, "a peer opens past the viewer stream limit"); + assert!(peer.max_data > 0, "a peer's wider receive window is advertised"); + + let viewer = probe(port, false).await; + assert!( + viewer.streams < PROBE && viewer.streams as u64 <= VIEWER_STREAMS + 1, + "a viewer keeps the stream limit, opened {}", + viewer.streams + ); + assert_eq!(viewer.max_data, 0, "a viewer's receive window is unchanged"); + + let _leases = deciding.await.expect("decider"); + running.abort(); + let _ = running.await; +} + +#[tokio::test] +async fn tokio_peer_session_widens_quic_limits() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let dir = tempfile::tempdir().expect("tempdir"); + let relay = Relay::load(config(dir.path())).await.expect("load relay"); + let port = relay.quic_addr().expect("quic listener bound").port(); + peer_widens_and_viewer_does_not(relay, port).await; +} + +/// Load a relay with `set` applied, returning why it was refused. +async fn refused_at_load(set: impl FnOnce(&mut Config)) -> String { + let dir = tempfile::tempdir().expect("tempdir"); + let mut config = config(dir.path()); + set(&mut config); + match Relay::load(config).await { + Ok(_) => panic!("an unadvertisable QUIC limit must be refused"), + Err(err) => format!("{err:#}"), + } +} + +/// A limit QUIC cannot advertise 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(); + let err = refused_at_load(|config| { + config.peer.quic.receive_window = std::num::NonZeroU64::new(1 << 62).unwrap(); + }) + .await; + assert!(err.contains("[peer.quic]") && err.contains("receive_window"), "{err}"); + + let err = refused_at_load(|config| { + config.peer.quic.max_streams = std::num::NonZeroU64::new((1 << 60) + 1).unwrap(); + }) + .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}"); +} + +/// A bare QUIC server speaking the dial template's ALPNs: the dialer's stream +/// limit is a transport parameter, so the handshake alone shows it. Its idle +/// timeout is far past any test's, so only a close ends a link in time. +pub(crate) fn bare_server(connect: &moq_tokio::connect::Config) -> driver::Endpoint { + let key = rcgen::KeyPair::generate().expect("keypair"); + let cert = rcgen::CertificateParams::new(vec!["localhost".to_string()]) + .expect("cert params") + .self_signed(&key) + .expect("self-signed cert"); + let mut tls = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert( + vec![cert.der().clone()], + rustls::pki_types::PrivateKeyDer::Pkcs8(key.serialize_der().into()), + ) + .expect("server tls"); + tls.alpn_protocols = connect + .versions() + .alpns() + .iter() + .map(|alpn| alpn.as_bytes().to_vec()) + .collect(); + let crypto = driver::crypto::rustls::QuicServerConfig::try_from(tls).expect("quic crypto"); + let mut transport = driver::TransportConfig::default(); + transport.max_idle_timeout(Some(BARE_IDLE.try_into().expect("idle timeout"))); + let mut config = driver::ServerConfig::with_crypto(std::sync::Arc::new(crypto)); + config.transport_config(std::sync::Arc::new(transport)); + driver::Endpoint::server(config, "127.0.0.1:0".parse().unwrap()).expect("server endpoint") +} + +/// Start a cluster whose one static peer is `server`, and accept its dial. +async fn static_dial( + server: &driver::Endpoint, + connect: moq_tokio::connect::Config, + quic: moq_tokio::quic::Config, +) -> (driver::Connection, tokio::task::JoinHandle>) { + let port = server.local_addr().expect("server addr").port(); + let mut options = moq_relay::cluster::Options::new(Default::default()); + options.config.connect = vec![moq_relay::cluster::Peer::new(format!("moql://127.0.0.1:{port}/"))]; + options.peer.quic.max_streams = std::num::NonZeroU64::new(4 * PROBE as u64).unwrap(); + let cluster = moq_relay::cluster::Cluster::new(options) + .expect("cluster init") + .with_connect(connect, quic); + let running = tokio::spawn(cluster.start().await.expect("cluster start").run()); + let conn = tokio::time::timeout(TIMEOUT, server.accept()) + .await + .expect("dial timeout") + .expect("incoming") + .await + .expect("handshake"); + (conn, running) +} + +/// Every cluster dial is a peer link, so a dial template carrying the plain +/// `[quic]` settings still advertises the peer limits from the handshake on. +#[tokio::test] +async fn cluster_dials_widen_a_plain_client() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let mut connect = moq_tokio::connect::Config::default(); + connect.tls.insecure = Some(true); + let server = bare_server(&connect); + let mut quic = moq_tokio::quic::Config::default(); + quic.max_streams = Some(VIEWER_STREAMS); + let (conn, running) = static_dial(&server, connect, quic).await; + + let mut streams = Vec::new(); + while streams.len() < PROBE { + match conn.open_uni().now_or_never() { + Some(stream) => streams.push(stream.expect("open")), + None => break, + } + } + assert_eq!(streams.len(), PROBE, "the dial advertises the peer stream limit"); + + running.abort(); + let _ = running.await; + conn.close(driver::VarInt::from_u32(0), b"done"); +} + +/// A static peer dials from the `[connect]` bind port, so a deployment that +/// admits cluster traffic by source port keeps working. +#[tokio::test] +async fn static_peer_dials_from_the_configured_bind() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let mut connect = moq_tokio::connect::Config::default(); + connect.tls.insecure = Some(true); + // Borrow a free port from the OS, then hand it to the dial template. + let bind = std::net::UdpSocket::bind("127.0.0.1:0") + .expect("probe socket") + .local_addr() + .expect("probe addr"); + connect.bind = Some(bind); + let server = bare_server(&connect); + let (conn, running) = static_dial(&server, connect, Default::default()).await; + + assert_eq!(conn.remote_address(), bind, "the dial keeps the configured bind"); + + running.abort(); + let _ = running.await; + conn.close(driver::VarInt::from_u32(0), b"done"); +} + +/// The tokio worker shards accept through the same request path, so a peer +/// widens there too. +#[cfg(target_os = "linux")] +#[tokio::test] +async fn tokio_worker_peer_session_widens_quic_limits() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let dir = tempfile::tempdir().expect("tempdir"); + let mut config = config(dir.path()); + config.runtime.workers = Some(1); + config.runtime.pin = false; + let relay = Relay::load(config).await.expect("load relay"); + let port = relay.addr().expect("workers bound an address").port(); + peer_widens_and_viewer_does_not(relay, port).await; +} + +#[cfg(all(target_os = "linux", feature = "_uring"))] +#[tokio::test] +async fn uring_peer_session_widens_quic_limits() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + match moq_uring::Worker::new(Default::default()) { + Ok(_) => {} + Err(moq_uring::Error::Unsupported(reason)) if std::env::var_os("MOQ_STRICT").is_none() => { + eprintln!("skipping io_uring relay test: {reason}"); + return; + } + Err(err) => panic!("io_uring worker setup failed: {err}"), + } + let dir = tempfile::tempdir().expect("tempdir"); + let mut config = config(dir.path()); + config.runtime.workers = Some(1); + config.runtime.pin = false; + config.runtime.io_uring = true; + let relay = Relay::load(config).await.expect("load relay"); + let port = relay.addr().expect("workers bound an address").port(); + peer_widens_and_viewer_does_not(relay, port).await; +} diff --git a/rs/moq-relay/tests/integration/upstream_links.rs b/rs/moq-relay/tests/integration/upstream_links.rs index 2d08a0270af..bef034a7554 100644 --- a/rs/moq-relay/tests/integration/upstream_links.rs +++ b/rs/moq-relay/tests/integration/upstream_links.rs @@ -98,14 +98,13 @@ async fn node(id: u64, connect: Vec) -> Node { let mut client = moq_tokio::connect::Config::default(); client.tls.insecure = Some(true); - let client = client.init(Default::default()).expect("client init"); let mut config = cluster::Config::default(); config.id = Some(id); config.connect = connect; let cluster = cluster::Cluster::new(cluster::Options::new(config)) .expect("cluster init") - .with_client(client); + .with_connect(client, Default::default()); let (auth, mut admissions) = auth::Auth::embedded(format!("node-{id}")); tasks.spawn(async move { diff --git a/rs/moq-tokio/src/error.rs b/rs/moq-tokio/src/error.rs index 7cca9e4b60a..497bb9a7ab8 100644 --- a/rs/moq-tokio/src/error.rs +++ b/rs/moq-tokio/src/error.rs @@ -105,6 +105,14 @@ 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")] + 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, @@ -251,6 +259,8 @@ impl Error { | Self::Deprecated(_) | Self::IdleTimeoutRange | Self::WindowRange(_) + | Self::StreamsRange + | Self::QuicLimits(_) | Self::QlogUnsupported | Self::Connect(_) => true, #[cfg(feature = "quic")] diff --git a/rs/moq-tokio/src/quic.rs b/rs/moq-tokio/src/quic.rs index f1e9d9fb739..a9938bf2224 100644 --- a/rs/moq-tokio/src/quic.rs +++ b/rs/moq-tokio/src/quic.rs @@ -110,8 +110,7 @@ pub const DEFAULT_MAX_STREAMS: u64 = 10_000; /// 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 = "quic", feature = "iroh"))] -pub(crate) const DEFAULT_RECEIVE_WINDOW: u64 = 64 << 20; +pub const DEFAULT_RECEIVE_WINDOW: u64 = 64 << 20; /// Default idle timeout before an inactive connection is dropped. /// @@ -126,6 +125,33 @@ 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)] @@ -532,9 +558,10 @@ impl Config { self.legacy.deprecated() } - /// Reject knobs this build can't honor. Called when the endpoint is built. + /// Reject knobs this build can't honor, including limits QUIC cannot + /// advertise. Called when the endpoint is built. #[cfg(feature = "_transport")] - pub(crate) fn validate(&self) -> crate::Result<()> { + pub fn validate(&self) -> crate::Result<()> { // Erroring beats silently ignoring the flag: the operator asked for traces and // would otherwise go looking for files that were never going to appear. match self.qlog.as_ref() { @@ -542,6 +569,7 @@ impl Config { _ => Ok(()), }?; validate_idle_timeout(Some(self.resolve().idle_timeout))?; + validate_streams(self.max_streams)?; validate_windows(self) } @@ -599,9 +627,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. +const MAX_STREAMS: u64 = 1 << 60; + +/// Reject a stream limit a peer would refuse with FRAME_ENCODING_ERROR. +fn validate_streams(max_streams: Option) -> crate::Result<()> { + match max_streams { + Some(streams) if streams > MAX_STREAMS => Err(crate::Error::StreamsRange), + _ => Ok(()), + } +} + /// Reject a flow-control window no connection can honor. /// /// Zero is rejected everywhere: a window of zero credits nothing, so it would wedge @@ -981,6 +1020,38 @@ 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. + #[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")) + )); + } + #[test] fn window_flags_parse() { let quic = parse(&[ diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index a46c5e425c2..c52bc17b504 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -708,10 +708,11 @@ 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 (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 }; - Ok(Request { transport, url, identity, authority, link, kind: RequestKind::Quic(Box::new(request)), deadline, active }) + Ok(Request { transport, url, identity, authority, link, kind: RequestKind::Quic(Box::new(request)), quic, deadline, active }) }.boxed()); } } @@ -728,7 +729,18 @@ impl Server { let Accepted { session, url, identity, authority, link } = within(deadline, super::iroh::accept(_conn, alpns)).await?; let binding = match url { Some(_) => moq_net::transport::Binding::Uri, None => moq_net::transport::Binding::InBand }; let (request, deadline) = accept_request(&setup.server, session, binding, deadline).await?; - Ok(Request { transport: Transport::Iroh, url, identity, authority, link, kind: RequestKind::Iroh(Box::new(request)), deadline, active }) + Ok(Request { + transport: Transport::Iroh, + url, + identity, + authority, + link, + kind: RequestKind::Iroh(Box::new(request)), + #[cfg(feature = "quic")] + quic: None, + deadline, + active, + }) }.boxed()); } } @@ -998,6 +1010,8 @@ pub(crate) async fn accept_websocket( identity: options.identity, link: options.link, kind: RequestKind::Qmux(Box::new(request)), + #[cfg(feature = "quic")] + quic: None, deadline, active: options.active, }) @@ -1485,6 +1499,8 @@ async fn stream_request( identity: None, link, kind: RequestKind::Qmux(Box::new(request)), + #[cfg(feature = "quic")] + quic: None, deadline, active, }; @@ -1564,6 +1580,10 @@ 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`]. + #[cfg(feature = "quic")] + quic: Option, /// What remains of the handshake deadline, which also bounds [`Request::ok`]. deadline: Option, } @@ -1667,6 +1687,31 @@ 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)) + } + /// Accept the session and return its supervised driver task. /// /// Await the driver before classifying the session close so a panic cannot look like cancellation. diff --git a/rs/moq-uring/src/error.rs b/rs/moq-uring/src/error.rs index b4a43aa6631..5a871aa995b 100644 --- a/rs/moq-uring/src/error.rs +++ b/rs/moq-uring/src/error.rs @@ -11,6 +11,14 @@ 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 da60852433b..37f02e827a6 100644 --- a/rs/moq-uring/src/quic/driver/connection.rs +++ b/rs/moq-uring/src/quic/driver/connection.rs @@ -372,6 +372,29 @@ impl Connection { &self.handshake } + /// Change the connection's flow-control limits; 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); + self.shared.kick(); + Ok(()) + } + /// Returns connection statistics. pub fn stats(&self) -> moq_quic::ConnectionStats { self.shared.conn.borrow().stats() diff --git a/rs/moq-uring/src/quic/mod.rs b/rs/moq-uring/src/quic/mod.rs index 70e2b099644..32cfb5908be 100644 --- a/rs/moq-uring/src/quic/mod.rs +++ b/rs/moq-uring/src/quic/mod.rs @@ -161,6 +161,22 @@ 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. diff --git a/rs/moq-uring/src/quic/web.rs b/rs/moq-uring/src/quic/web.rs index 911d940887c..3800b0b817c 100644 --- a/rs/moq-uring/src/quic/web.rs +++ b/rs/moq-uring/src/quic/web.rs @@ -478,6 +478,12 @@ 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> { + self.conn.set_limits(limits) + } + /// Assemble the web flavor and spawn its capsule reader. fn establish(request: Request, protocol: Option) -> Self { let Request {