From 59a4ba85e08fb07842100e8fc5e2af6be82be481 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 01:50:24 -0800 Subject: [PATCH 1/8] chore(quest): claim quest/m2/relay-quic-lb-forward.md Quest-Claim: 50c02e22f439093a From 8517522fb6b37704fe72aa087cde0c99d52978e4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 02:11:42 -0800 Subject: [PATCH 2/8] feat(tokio)!: forward QUIC packets to their owner by QUIC-LB server ID Co-Authored-By: Claude Opus 5.5 --- doc/bin/relay/config.md | 34 ++ quest/m2/README.md | 1 - quest/m2/relay-quic-lb-forward.md | 69 ---- rs/moq-relay/src/uring.rs | 7 +- rs/moq-tokio/Cargo.toml | 2 +- rs/moq-tokio/src/listen.rs | 138 ++++++++ rs/moq-tokio/src/quic.rs | 16 +- rs/moq-tokio/src/quic/backend.rs | 157 +++++++-- rs/moq-tokio/src/quic/driver/endpoint.rs | 148 ++++++-- rs/moq-tokio/src/quic/driver/forward.rs | 410 +++++++++++++++++++++++ rs/moq-tokio/src/quic/driver/mod.rs | 1 + rs/moq-tokio/src/quic/driver/tests.rs | 106 +++++- 12 files changed, 959 insertions(+), 130 deletions(-) delete mode 100644 quest/m2/relay-quic-lb-forward.md create mode 100644 rs/moq-tokio/src/quic/driver/forward.rs diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 3710eba5161..74cea22b5eb 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -24,6 +24,15 @@ implementation = "edge/1" # SETUP name for the relay. Omit for moq-re [listen.reset] secret = "/etc/moq/reset.key" # File holding 64 hex chars. Omit for a random key. +[listen.lb] # QUIC-LB: name this relay in its connection IDs. +id = "01" # Server ID, hex. The same length on every relay. +nonce = 8 # Random bytes after it. Default. + +[listen.lb.forward] # Forward packets that land on the wrong relay to their owner. +bind = "10.0.0.1:4443" # This relay's forward socket, a specific IP its peers list. +key = "/etc/moq/forward.key" # File holding 64 hex chars, the same on every relay. +peers = "/etc/moq/peers.toml" # Server ID to forward socket. Reloaded on change. + [listen.tls] cert = "cert.pem" # Certificate chain and key. Reloaded on change. key = "key.pem" @@ -72,6 +81,31 @@ The io\_uring runtime refuses `preferred_v4`, `preferred_v6`, and `lb.id`; use the tokio runtime when the listener needs these settings. `[listen.lb] nonce` needs an `id` in the same table; unset nonce uses 8. The `--listen-quic-lb-id` and `--listen-quic-lb-nonce` flags update that table. + +`[listen.lb.forward]` keeps a connection alive when its packets land on another relay behind +the same anycast address or load-balanced VIP, after a route change or a rebalance. That relay +reads the server ID in the packet's connection ID and forwards the packet to the owner's +forward socket, which processes it as if the client had sent it there. The owner keeps replying +straight to the client from the shared address its handshake arrived on, so only the +client-to-server direction takes the extra hop. A browser never leaves that address: Chrome's +WebTransport ignores the QUIC `preferred_address`. Handshakes are not forwarded, so the front +end must keep each new flow on one relay until it completes; a split handshake fails and the +client retries. + +The peers file maps every relay's server ID to its forward `bind`, and may list this relay too, +so one file can serve the whole cluster: + +```toml +"01" = "10.0.0.1:4443" +"02" = "10.0.0.2:4443" +``` + +A packet for a server ID missing from it is dropped. A forward is accepted only from a listed +address and under the shared `key` (`openssl rand -hex 32 > forward.key`), since it carries a +client address the owner takes on trust. A changed file replaces the map; an invalid one is +logged and the last valid map kept, while an invalid one at startup stops it. Forwarding needs +`lb.id` and all three settings, and the +`--listen-quic-lb-forward-bind`, `-key`, and `-peers` flags update that table. `unix.allow` needs `unix.bind`. ## \[quic] diff --git a/quest/m2/README.md b/quest/m2/README.md index 2dfe86b62d4..f811ebf53b5 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -32,7 +32,6 @@ lite-07; bitrate enforcement is not part of this release. - [Weekly mutation testing on critical modules](/quest/m2/mutants.md) - cargo-mutants runs weekly on moq-auth and moq-net's permission, resume, and expiry code - [Video quality](/quest/m2/video-quality/README.md) - a nightly VMAF harness grades every encoder backend and on-demand GoP pops against floors, and moq-boy's encode is fixed under it - [noq adopts moq-congestion](/quest/m2/noq-congestion.md) - noq's BBR3 and Cubic delegate to the published crate behind noq's unchanged API -- [moq-relay forwards QUIC packets by QUIC-LB server ID](/quest/m2/relay-quic-lb-forward.md) - an established connection that lands on the wrong instance behind an anycast IP or VIP is forwarded to its owner, which replies from the shared address - [P2P](/quest/m2/p2p/README.md) - application-selected peer sessions use shared priorities, routing, and same-epoch handover - [rust2 publishes its UniFFI fork](/quest/m2/rust2-uniffi.md) - waiting on moq-dev/rust2 to import UniFFI and moq's generators and publish them - [Switch to rust2's UniFFI](/quest/m2/uniffi-fork.md) - moq-ffi and every wrapper build from moq-dev/rust2's fork under one pin, and moq's generator code is deleted diff --git a/quest/m2/relay-quic-lb-forward.md b/quest/m2/relay-quic-lb-forward.md deleted file mode 100644 index 7c4476ee0a1..00000000000 --- a/quest/m2/relay-quic-lb-forward.md +++ /dev/null @@ -1,69 +0,0 @@ -# [L] moq-relay forwards QUIC packets by QUIC-LB server ID - -## Goal - -A moq-relay QUIC listener with a QUIC-LB server ID and a peer map forwards -every short-header packet whose connection ID names another server to that -server's unicast address, carrying the client's address. The owning relay -processes it as if it came from the client, and sends that connection's -packets with a configured return source address (direct server return), so -an established connection survives landing on the wrong instance behind an -anycast IP or a load-balancing VIP. Long-header packets are never forwarded: -the handshake relies on the front end keeping a flow on one instance. The -flags are documented inline, with MIGRATE notes. - -## Plan - -Settled with the maintainer, 2026-10-30, for moq.pro's anycast edge cluster -on Cloudflare: - -- Why: Chrome's WebTransport never follows `preferred_address`. Chromium - sends every `https` WebTransport through `DedicatedWebTransportHttp3Client`, - a bare `QuicSpdyClientSession` with no migration path, while its pooled - HTTP/3 client validates the server preferred address. Measured locally - against a moq-relay advertising `--listen-preferred-v4` (Chrome 156: the - net-log shows the address received and never acted on; a forced-QUIC - `fetch()` sends `PATH_CHALLENGE` to it). moq-cli does not migrate either. - So a browser stays on the anycast or VIP address for a connection's life, - and a route change or a VIP rebalance lands its packets on another - instance. -- Scope: established connections only. A handshake split across instances - fails and the client retries. Rejected: a Retry carrying the server ID - (one extra round trip per connection) and a consistent hash of the - client's ID (needs an identical membership view everywhere). -- Connection IDs: the existing plaintext QUIC-LB generator (`listen.lb.id`, - `listen.lb.nonce`, [#848](https://github.com/moq-dev/moq/pull/848)), not the - draft's encrypted form. -- Return path: direct server return. The owner sends a forwarded - connection's packets with the configured return address (the anycast or - VIP address) as source, so only the client-to-server direction takes the - extra hop. Whether the platform allows that source is the deployment's - concern; moq.pro asks Cloudflare. -- Peer map: server ID to unicast address, updated at runtime (the deployment - feeds it, moq.pro from its own API), through config the relay hot-reloads - or the library API; the relay refuses a forward to an unknown ID. - -Open for implementation: - -- The forwarding encapsulation: the original client address and the - address the packet arrived on, prepended to the datagram and sent to the - peer's QUIC port or a dedicated one. Forwards are accepted only from - peer-map addresses and authenticated (a shared key), since a forged - client address would let anyone inject packets or reflect traffic. -- Sending from the return address: a socket that may use a non-local - source (`IP_FREEBIND` or `IP_TRANSPARENT` on Linux) versus binding the - return address where the platform routes it. -- Tests: three relays on loopback addresses, a connection established on - one, its short-header packets delivered to another, and the session - continuing. - -Interactions: per-core shard steering already refuses QUIC-LB connection -IDs (`ShardWithQuicLb`), so if -[Shard the relay by default](/quest/m2/perf/shard-default.md) flips the -default, a listener with `lb.id` stays unsharded. The io_uring listener's -`lb.id` refusal ([io_uring package](/quest/m2/relay-io-uring-package.md)) -can stand: forwarding lives on the tokio backend unless a consumer asks. - -Public API: new listener settings (peer map, return address, forward key) -and a library entry point to update the peer map. Wire: none toward clients; -the peer-to-peer forward format is internal. diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index 217e2432788..a1e44dd409f 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -147,7 +147,12 @@ impl Workers { tls: listen_tls, preferred_v4, preferred_v6, - lb: moq_tokio::listen::LoadBalancer { id: lb_id, nonce: _ }, + lb: moq_tokio::listen::LoadBalancer { + id: lb_id, + nonce: _, + // Needs an id, refused below. + forward: _, + }, reset_secret, } = server.listen.resolve()?; let quic = &server.quic; diff --git a/rs/moq-tokio/Cargo.toml b/rs/moq-tokio/Cargo.toml index fe5b46194a1..656418b91a8 100644 --- a/rs/moq-tokio/Cargo.toml +++ b/rs/moq-tokio/Cargo.toml @@ -19,7 +19,7 @@ default = ["net", "quic", "aws-lc-rs", "websocket", "tcp", "uds"] net = ["dep:moq-net", "dep:web-transport-trait"] # The async QUIC driver alone (`quic::driver`): moq-quic on tokio, without moq-net or # the MoQ backend. Needs a crypto-provider feature. -driver = ["dep:moq-congestion", "dep:moq-quic", "dep:pin-project-lite", "dep:rustc-hash"] +driver = ["dep:hmac", "dep:moq-congestion", "dep:moq-quic", "dep:pin-project-lite", "dep:rustc-hash", "dep:sha2"] # QUIC and WebTransport over moq-quic, for `https://`, `moqt://`, and `moql://`. quic = ["_transport", "_certs", "driver", "dep:http", "dep:web-transport-proto", "dep:reqwest", "watch", "dep:dns-lookup"] # Server-side certificate handling: chains loaded from disk, self-signed generation, diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs index a5d25359d17..9d3da04f79b 100644 --- a/rs/moq-tokio/src/listen.rs +++ b/rs/moq-tokio/src/listen.rs @@ -447,6 +447,34 @@ impl Config { "--listen-quic-lb-nonce requires --listen-quic-lb-id", )); } + let forward = &self.lb.forward; + if forward.configured() { + let refused = [ + ( + self.lb.id.is_none(), + "--listen-quic-lb-forward-* requires --listen-quic-lb-id", + ), + ( + forward.bind.is_none(), + "--listen-quic-lb-forward-* requires --listen-quic-lb-forward-bind", + ), + ( + forward.key.is_none(), + "--listen-quic-lb-forward-bind requires --listen-quic-lb-forward-key", + ), + ( + forward.peers.is_none(), + "--listen-quic-lb-forward-bind requires --listen-quic-lb-forward-peers", + ), + ( + forward.bind.is_some_and(|bind| bind.ip().is_unspecified()), + "--listen-quic-lb-forward-bind needs a specific IP, which its peers list", + ), + ]; + if let Some((_, reason)) = refused.into_iter().find(|(set, _)| *set) { + return Err(crate::Error::NoBackend(reason)); + } + } Ok(()) } @@ -562,9 +590,18 @@ impl Config { #[cfg(feature = "quic")] /// Return the effective QUIC-LB connection-ID encoding. pub(crate) fn load_balancer(&self) -> Option { + let forward = &self.lb.forward; self.lb.id.clone().map(|id| crate::quic::LoadBalancer { id, nonce: self.lb.nonce.unwrap_or(8), + forward: match (forward.bind, &forward.key, &forward.peers) { + (Some(bind), Some(key), Some(peers)) => Some(crate::quic::Forward { + bind, + key: key.clone(), + peers: peers.clone(), + }), + _ => None, + }, }) } } @@ -706,6 +743,7 @@ mod tests { Some(crate::quic::LoadBalancer { id: "ab".parse().unwrap(), nonce: 9, + forward: None, }) ); } @@ -741,6 +779,11 @@ mod tests { [lb] id = "ab" nonce = 8 + +[lb.forward] +bind = "10.0.0.1:4443" +key = "forward.key" +peers = "peers.toml" "#, ) .unwrap(); @@ -749,10 +792,43 @@ nonce = 8 Some(crate::quic::LoadBalancer { id: "ab".parse().unwrap(), nonce: 8, + forward: Some(crate::quic::Forward { + bind: "10.0.0.1:4443".parse().unwrap(), + key: "forward.key".into(), + peers: "peers.toml".into(), + }), }) ); } + /// Forwarding needs a server ID and all three of its settings, and a bind its peers can name. + #[test] + fn quic_lb_forward_refuses_a_partial_config() { + let complete = "[lb]\nid = \"ab\"\n[lb.forward]\nbind = \"10.0.0.1:4443\"\nkey = \"k\"\npeers = \"p\"\n"; + let config: Config = toml::from_str(complete).unwrap(); + config.validate().expect("complete"); + + for (body, needle) in [ + (complete.replace("id = \"ab\"\n", ""), "requires --listen-quic-lb-id"), + ( + complete.replace("bind = \"10.0.0.1:4443\"\n", ""), + "requires --listen-quic-lb-forward-bind", + ), + (complete.replace("key = \"k\"\n", ""), "requires --listen-quic-lb-forward-key"), + ( + complete.replace("peers = \"p\"\n", ""), + "requires --listen-quic-lb-forward-peers", + ), + (complete.replace("10.0.0.1", "0.0.0.0"), "needs a specific IP"), + ] { + let config: Config = toml::from_str(&body).unwrap(); + match config.validate() { + Err(crate::Error::NoBackend(reason)) => assert!(reason.contains(needle), "{body}: {reason}"), + other => panic!("{body}: {other:?}"), + } + } + } + /// A stream-only server opens no QUIC listener even with a bind configured, /// which is what lets a worker group hold that address instead. #[tokio::test] @@ -902,4 +978,66 @@ pub struct LoadBalancer { )] #[serde(default, skip_serializing_if = "Option::is_none")] pub nonce: Option, + + /// Forwarding to the other servers behind this one's address (`[listen.lb.forward]`). + #[usage(flatten)] + #[serde(default)] + pub forward: Forward, +} + +/// QUIC-LB forwarding (`[listen.lb.forward]`, `--listen-quic-lb-forward-*`). +/// +/// Behind an anycast address or a load-balanced VIP, a route change or a rebalance can land an +/// established connection's packets on another server. With forwarding, that server hands each +/// packet whose connection ID names another server to the owner, which processes it as if the +/// client had sent it there and keeps replying from the shared address. Only packets after the +/// handshake are forwarded, so the front end must still keep a handshake on one server. +#[derive(Clone, Debug, Default, usage_rs::Args, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +#[non_exhaustive] +pub struct Forward { + /// The UDP address this server receives forwards on, and the one its peers name it by. + /// Needs a specific IP, since a forward is accepted only from a peer's listed address. + #[usage( + name = "listen-quic-lb-forward-bind", + long = "listen-quic-lb-forward-bind", + env = "MOQ_LISTEN_QUIC_LB_FORWARD_BIND", + setting = "listen.lb.forward.bind" + )] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bind: Option, + + /// A file holding the key every server forwarding to this one shares, as 64 hexadecimal + /// characters (`openssl rand -hex 32`). It authenticates each forward, which carries a + /// client address the receiver would otherwise take on trust. + #[usage( + name = "listen-quic-lb-forward-key", + long = "listen-quic-lb-forward-key", + env = "MOQ_LISTEN_QUIC_LB_FORWARD_KEY", + setting = "listen.lb.forward.key" + )] + #[usage(value_hint = usage_rs::ValueHint::FilePath)] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub key: Option, + + /// A TOML file naming each server's forward address by its hex server ID, such as + /// `"02" = "10.0.0.2:4443"`. Reloaded when it changes, keeping the last valid map. + /// This server's own entry is skipped, so every server can share one file. A packet for + /// a server missing from it is dropped. + #[usage( + name = "listen-quic-lb-forward-peers", + long = "listen-quic-lb-forward-peers", + env = "MOQ_LISTEN_QUIC_LB_FORWARD_PEERS", + setting = "listen.lb.forward.peers" + )] + #[usage(value_hint = usage_rs::ValueHint::FilePath)] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub peers: Option, +} + +impl Forward { + /// Whether any forwarding setting is present. + pub fn configured(&self) -> bool { + self.bind.is_some() || self.key.is_some() || self.peers.is_some() + } } diff --git a/rs/moq-tokio/src/quic.rs b/rs/moq-tokio/src/quic.rs index 1aece9d47a0..3684e58cace 100644 --- a/rs/moq-tokio/src/quic.rs +++ b/rs/moq-tokio/src/quic.rs @@ -35,7 +35,7 @@ pub mod web; /// Parsed from, and serialized as, a hex string. Its length must match the load /// balancer's configured server-ID length. #[serde_with::serde_as] -#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[derive(Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)] pub struct ServerId(#[serde_as(as = "serde_with::hex::Hex")] pub(crate) Vec); impl ServerId { @@ -67,6 +67,20 @@ pub(crate) struct LoadBalancer { pub id: ServerId, /// Number of random nonce bytes appended to each connection ID. pub nonce: usize, + /// Forwarding to the other servers behind this one's address. + pub forward: Option, +} + +/// Resolved QUIC-LB forwarding, every setting present. +#[cfg(feature = "quic")] +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct Forward { + /// The forward socket's address. + pub bind: std::net::SocketAddr, + /// The shared key file. + pub key: PathBuf, + /// The peer map file. + pub peers: PathBuf, } /// The congestion control family for a QUIC connection. diff --git a/rs/moq-tokio/src/quic/backend.rs b/rs/moq-tokio/src/quic/backend.rs index 7fe873b77a3..7dec3ac467e 100644 --- a/rs/moq-tokio/src/quic/backend.rs +++ b/rs/moq-tokio/src/quic/backend.rs @@ -239,6 +239,27 @@ pub enum Error { /// The flow-control limits are ones a peer would refuse, so none were applied. #[error("invalid quic limits: {0}")] Limits(#[from] moq_quic::LimitsError), + + /// The QUIC-LB forward socket failed to bind. + #[error("failed to bind the QUIC-LB forward socket")] + BindForward(#[source] std::io::Error), + + /// The QUIC-LB forward key file could not be read or does not hold 64 hexadecimal characters. + #[error("invalid QUIC-LB forward key")] + ForwardKey(#[source] crate::secret::Error), + + /// The QUIC-LB peers file could not be read, or names a peer this server cannot forward to. + #[error("invalid QUIC-LB peers file {}: {reason}", path.display())] + ForwardPeers { + /// The peers file. + path: std::path::PathBuf, + /// What is wrong with it. + reason: String, + }, + + /// The QUIC-LB peers file could not be watched for changes. + #[error("failed to watch the QUIC-LB peers file: {0}")] + WatchPeers(String), } impl crate::failover::Aggregate for Error { @@ -554,6 +575,8 @@ fn proto_status(err: &web::proto::ConnectError) -> Option { pub(crate) struct Server { pub quic: driver::endpoint::Endpoint, tls: crate::tls::Serve, + /// Reloads the QUIC-LB peers for as long as the server runs. + _peers: Option, } pub(crate) struct Options { @@ -619,6 +642,7 @@ impl Server { endpoint_config.reset_secret = Some(*secret.as_bytes()); } + let mut forward = None; // Configure connection ID generator with server ID if provided if let Some(shard) = member.as_ref().map(listen::Socket::shard) { if load_balancer.is_some() { @@ -653,8 +677,15 @@ impl Server { nonce_len, "using QUIC-LB compatible connection ID generation" ); + if let Some(config) = &load_balancer.forward { + forward = Some(Forwarding::start( + config, + &server_id, + endpoint_config.max_udp_payload_size, + )?); + } endpoint_config.cid_generator = Some(Arc::new(move || { - Box::new(ServerIdGenerator::new(server_id.clone(), nonce_len)) + Box::new(driver::forward::ServerIdGenerator::new(&server_id.0, nonce_len)) })); } @@ -665,9 +696,20 @@ impl Server { None => crate::bind::udp(crate::bind::Udp::new(listen)).map_err(Error::BindSocket)?, }; - let quic = driver::endpoint::Endpoint::new(endpoint_config, socket).map_err(Error::CreateEndpoint)?; + let (quic, peers) = match forward { + Some(Forwarding { forward, watcher }) => ( + driver::endpoint::Endpoint::forwarding(endpoint_config, socket, forward), + Some(watcher), + ), + None => (driver::endpoint::Endpoint::new(endpoint_config, socket), None), + }; + let quic = quic.map_err(Error::CreateEndpoint)?; - Ok(Self { quic, tls: served }) + Ok(Self { + quic, + tls: served, + _peers: peers, + }) } pub fn accept(&self) -> impl std::future::Future> + '_ { @@ -811,38 +853,68 @@ pub(crate) async fn accept( } } -// ── ServerIdGenerator ─────────────────────────────────────────────── +// ── Forwarding ────────────────────────────────────────────────────── -struct ServerIdGenerator { - server_id: ServerId, - nonce_len: usize, +/// A forward socket, and the watcher that keeps its peers current. +struct Forwarding { + forward: driver::forward::Forward, + watcher: notify::RecommendedWatcher, } -impl ServerIdGenerator { - fn new(server_id: ServerId, nonce_len: usize) -> Self { - Self { server_id, nonce_len } - } -} - -impl driver::ConnectionIdGenerator for ServerIdGenerator { - fn generate_cid(&mut self) -> driver::ConnectionId { - use rand::RngExt; - let cid_len = self.cid_len(); - let mut cid = Vec::with_capacity(cid_len); - // First byte has "self-encoded length" of server ID + nonce - cid.push((cid_len - 1) as u8); - cid.extend(self.server_id.0.iter()); - cid.extend(rand::rng().random_iter::().take(self.nonce_len)); - driver::ConnectionId::new(cid.as_slice()) - } +impl Forwarding { + fn start(config: &crate::quic::Forward, id: &ServerId, max_udp_payload_size: u16) -> Result { + let key = crate::secret::Secret::read(&config.key).map_err(Error::ForwardKey)?; + let peers = driver::forward::Peers::default(); + peers.set(read_peers(&config.peers, id, config.bind)?); + + let reload = peers.clone(); + let (path, own, bind) = (config.peers.clone(), id.clone(), config.bind); + let watcher = crate::watch::callback(std::slice::from_ref(&config.peers), move || { + match read_peers(&path, &own, bind) { + Ok(map) => { + reload.set(map); + tracing::info!(path = %path.display(), "reloaded QUIC-LB peers"); + } + Err(err) => tracing::warn!(%err, "failed to reload QUIC-LB peers; retaining the previous ones"), + } + }) + .map_err(|err| Error::WatchPeers(err.to_string()))?; - fn cid_len(&self) -> usize { - 1 + self.server_id.len() + self.nonce_len + let socket = std::net::UdpSocket::bind(config.bind).map_err(Error::BindForward)?; + let forward = driver::forward::Forward::new(socket, &id.0, peers, key.as_bytes(), max_udp_payload_size) + .map_err(Error::BindForward)?; + tracing::info!(bind = %config.bind, "forwarding QUIC-LB packets"); + Ok(Self { forward, watcher }) } +} - fn cid_lifetime(&self) -> Option { - None +/// Read the peers file, skipping this server's own entry. +/// +/// Refuses the whole file for one bad entry, so a typo cannot quietly strand that server's +/// clients: a server ID of another length, which this server's connection IDs could not carry, +/// or an address it cannot send to or accept from. +fn read_peers(path: &std::path::Path, own: &ServerId, bind: net::SocketAddr) -> Result { + let invalid = |reason: String| Error::ForwardPeers { + path: path.to_owned(), + reason, + }; + let text = std::fs::read_to_string(path).map_err(|err| invalid(err.to_string()))?; + let peers: std::collections::HashMap = + toml::from_str(&text).map_err(|err| invalid(err.to_string()))?; + + let mut map = driver::forward::Map::default(); + for (id, addr) in peers { + if id.len() != own.len() { + return Err(invalid(format!("{id:?} is not {} bytes like this server's ID", own.len()))); + } + if addr.ip().is_unspecified() || addr.is_ipv4() != bind.is_ipv4() { + return Err(invalid(format!("{id:?} has an address {bind} cannot forward to: {addr}"))); + } + if id != *own { + map.insert(&id.0, addr); + } } + Ok(map) } #[cfg(test)] @@ -1061,6 +1133,35 @@ mod tests { .expect("test timed out"); } + /// The peers file skips this server's own entry, so every server can share it, and refuses + /// the whole file for one entry this server could not forward to. + #[test] + fn peers_file_is_shared_and_refused_whole() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("peers.toml"); + let own: ServerId = "01".parse().unwrap(); + let bind: net::SocketAddr = "10.0.0.1:4443".parse().unwrap(); + let read = |body: &str| { + std::fs::write(&path, body).unwrap(); + read_peers(&path, &own, bind) + }; + + let map = read("\"01\" = \"10.0.0.1:4443\"\n\"02\" = \"10.0.0.2:4443\"\n").unwrap(); + let expected: std::collections::HashMap, net::SocketAddr> = + [(Box::from(&[2u8][..]), "10.0.0.2:4443".parse().unwrap())].into(); + assert_eq!(map.peers(), &expected); + + for (body, needle) in [ + ("\"0203\" = \"10.0.0.2:4443\"\n", "not 1 bytes"), + ("\"02\" = \"[2001:db8::2]:4443\"\n", "cannot forward to"), + ("\"02\" = \"0.0.0.0:4443\"\n", "cannot forward to"), + ("\"zz\" = \"10.0.0.2:4443\"\n", "Invalid character"), + ] { + let err = read(body).expect_err(body).to_string(); + assert!(err.contains(needle), "{body}: {err}"); + } + } + /// Whether a live connection is running BBRv3. fn is_bbr3(conn: &driver::connection::Connection) -> bool { conn.congestion_state() diff --git a/rs/moq-tokio/src/quic/driver/endpoint.rs b/rs/moq-tokio/src/quic/driver/endpoint.rs index e12c2286739..8f560e54064 100644 --- a/rs/moq-tokio/src/quic/driver/endpoint.rs +++ b/rs/moq-tokio/src/quic/driver/endpoint.rs @@ -31,6 +31,7 @@ pub use crate::quic::driver::runtime::AsyncUdpSocket; use crate::quic::driver::{ ConnectionEvent, IO_LOOP_BOUND, Instant, RECV_TIME_BOUND, VarInt, connection::Connecting, + forward::Forward, runtime::{Slot, wrap_udp_socket}, udp_transmit, work_limiter::WorkLimiter, @@ -99,8 +100,19 @@ impl Endpoint { /// /// The endpoint's tasks run on the caller's tokio runtime, else moq-tokio's fallback. pub fn new(config: Config, socket: impl Into) -> Result { + Self::start(config, socket.into(), None) + } + + /// [`new`](Self::new), forwarding each packet whose QUIC-LB server ID names another server + /// and accepting the packets those servers forward here. + #[cfg_attr(not(feature = "quic"), expect(dead_code, reason = "only the QUIC backend forwards"))] + pub(crate) fn forwarding(config: Config, socket: impl Into, forward: Forward) -> Result { + Self::start(config, socket.into(), Some(forward)) + } + + fn start(config: Config, socket: Socket, forward: Option) -> Result { let runtime = crate::runtime::handle(); - let socket = socket.into().into_async()?; + let socket = socket.into_async()?; let addr = socket.local_addr()?; let allow_mtud = !socket.may_fragment(); let max_receive_segments = socket.max_receive_segments(); @@ -112,6 +124,7 @@ impl Endpoint { max_udp_payload_size, max_receive_segments, runtime.clone(), + forward, ); let driver = EndpointDriver(rc.clone()); runtime.spawn( @@ -502,6 +515,8 @@ pub(crate) struct State { driver_lost: bool, /// Connections in the process of being accepted pending_accepts: usize, + /// QUIC-LB forwarding to and from the other servers behind this one's address + forward: Option, } #[derive(Debug)] @@ -522,6 +537,7 @@ impl State { let poll_res = self.recv_state.poll_socket( cx, &mut self.inner, + self.forward.as_mut(), &mut |cx, bufs, meta| socket.poll_recv(cx, bufs, meta), slot, now, @@ -533,10 +549,19 @@ impl State { let poll_res = self.recv_state.poll_socket( cx, &mut self.inner, + self.forward.as_mut(), &mut |cx, bufs, meta| slot.poll_recv(cx, bufs, meta), slot, now, ); + // Polled even when the socket used up the budget, which still admits one forward, so a + // busy socket cannot starve them. + let forwards = match self.forward.as_mut() { + Some(forward) => self + .recv_state + .poll_forwards(cx, &mut self.inner, forward, self.ipv6, slot, now), + None => Ok(false), + }; self.recv_state.recv_limiter.finish_cycle(get_time); let poll_res = poll_res?; if poll_res.received_connection_packet { @@ -544,7 +569,7 @@ impl State { // one anymore. TODO: Account for multiple outgoing connections. self.prev_socket = None; } - Ok(poll_res.keep_going) + Ok(poll_res.keep_going || forwards?) } fn handle_events(&mut self, cx: &mut Context<'_>, shared: &Shared) -> bool { @@ -768,6 +793,7 @@ impl EndpointRef { max_udp_payload_size: u16, max_receive_segments: usize, runtime: tokio::runtime::Handle, + forward: Option, ) -> Self { let (sender, events) = mpsc::unbounded_channel(); let recv_state = RecvState::new(sender, max_receive_segments, max_udp_payload_size); @@ -792,6 +818,7 @@ impl EndpointRef { driver_lost: false, recv_state, pending_accepts: 0, + forward, }), })) } @@ -865,6 +892,7 @@ impl RecvState { &mut self, cx: &mut Context<'_>, endpoint: &mut moq_quic::endpoint::Endpoint, + mut forward: Option<&mut Forward>, recv: &mut PollRecv<'_>, slot: &Slot, now: Instant, @@ -903,37 +931,19 @@ impl RecvState { let mut data = self.datagrams.split_to(meta.len); while !data.is_empty() { let buf = data.split_to(meta.stride.min(data.len())); - let mut response_buffer = Vec::new(); + if let Some(forward) = forward.as_deref_mut() + && forward.route(&buf, meta.addr, meta.dst_ip, meta.ecn) + { + continue; + } let received = Received { remote: meta.addr, local_ip: meta.dst_ip, ecn: meta.ecn.map(proto_ecn), data: buf, }; - match endpoint.handle(now, received, &mut response_buffer) { - Some(DatagramEvent::NewConnection(incoming)) => { - if self.connections.close.is_none() { - self.incoming.push_back(incoming); - } else { - let transmit = endpoint.refuse(incoming, &mut response_buffer); - respond(transmit, &response_buffer, slot); - } - } - Some(DatagramEvent::ConnectionEvent(handle, event)) => { - // Ignoring errors from dropped connections that haven't yet been cleaned up - received_connection_packet = true; - let _ = self - .connections - .senders - .get_mut(&handle) - .unwrap() - .send(ConnectionEvent::Proto(event)); - } - Some(DatagramEvent::Response(transmit)) => { - respond(transmit, &response_buffer, slot); - } - None => {} - } + received_connection_packet |= + Self::handle(&mut self.incoming, &mut self.connections, endpoint, now, received, slot); } } } @@ -969,6 +979,89 @@ impl RecvState { } } +impl RecvState { + /// Hand one received datagram to the endpoint, returning whether it reached a connection. + fn handle( + incoming_queue: &mut VecDeque, + connections: &mut ConnectionSet, + endpoint: &mut moq_quic::endpoint::Endpoint, + now: Instant, + received: Received, + slot: &Slot, + ) -> bool { + let mut response_buffer = Vec::new(); + match endpoint.handle(now, received, &mut response_buffer) { + Some(DatagramEvent::NewConnection(incoming)) => { + if connections.close.is_none() { + incoming_queue.push_back(incoming); + } else { + let transmit = endpoint.refuse(incoming, &mut response_buffer); + respond(transmit, &response_buffer, slot); + } + } + Some(DatagramEvent::ConnectionEvent(handle, event)) => { + // Ignoring errors from dropped connections that haven't yet been cleaned up + let _ = connections + .senders + .get_mut(&handle) + .unwrap() + .send(ConnectionEvent::Proto(event)); + return true; + } + Some(DatagramEvent::Response(transmit)) => { + respond(transmit, &response_buffer, slot); + } + None => {} + } + false + } + + /// Process the packets other servers forwarded here, returning whether the work limiter + /// stopped before they ran out. + fn poll_forwards( + &mut self, + cx: &mut Context<'_>, + endpoint: &mut moq_quic::endpoint::Endpoint, + forward: &mut Forward, + ipv6: bool, + slot: &Slot, + now: Instant, + ) -> Result { + loop { + match forward.poll_recv(cx) { + Poll::Ready(Ok(Some(unwrapped))) => { + // The client's address as this socket would have reported it, so the + // connection sees the path its handshake took. An IPv6 client cannot have + // shaken hands on an IPv4 socket. + let remote = match (ipv6, unwrapped.remote) { + (true, remote) => Some(SocketAddr::V6(ensure_ipv6(remote))), + (false, remote @ SocketAddr::V4(_)) => Some(remote), + (false, SocketAddr::V6(_)) => None, + }; + if let Some(remote) = remote { + self.datagrams.extend_from_slice(unwrapped.packet); + let received = Received { + remote, + local_ip: unwrapped.local_ip, + ecn: unwrapped.ecn.map(proto_ecn), + data: self.datagrams.split(), + }; + Self::handle(&mut self.incoming, &mut self.connections, endpoint, now, received, slot); + } + } + Poll::Ready(Ok(None)) => {} + Poll::Pending => return Ok(false), + Poll::Ready(Err(e)) if e.kind() == io::ErrorKind::ConnectionReset || is_msg_size_err(&e) => {} + Poll::Ready(Err(e)) => return Err(e), + } + self.recv_limiter.record_work(1); + if !self.recv_limiter.allow_work(Instant::now) { + return Ok(true); + } + } + } +} + impl fmt::Debug for RecvState { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("RecvState") @@ -1027,6 +1120,7 @@ mod tests { .poll_socket( &mut cx, &mut endpoint, + None, &mut |cx, bufs, meta| socket.poll_recv(cx, bufs, meta), &slot, now, diff --git a/rs/moq-tokio/src/quic/driver/forward.rs b/rs/moq-tokio/src/quic/driver/forward.rs new file mode 100644 index 00000000000..f172af6bfce --- /dev/null +++ b/rs/moq-tokio/src/quic/driver/forward.rs @@ -0,0 +1,410 @@ +//! QUIC-LB forwarding between the servers behind one address. +//! +//! A server whose connection IDs carry a QUIC-LB server ID hands each short-header packet naming +//! another server to that server's forward socket, wrapped with the client's address and the +//! address the packet arrived on. The owner unwraps it and processes it as if the client had sent +//! it there. Replies need nothing more: a connection sends from the address its handshake arrived +//! on, which is the shared one, so only the client-to-server direction takes the extra hop. +//! +//! Long headers are never forwarded: a client picks the destination ID of its handshake, so the +//! handshake relies on the front end keeping a flow on one server. +#![cfg_attr(not(feature = "quic"), allow(dead_code, reason = "only the QUIC backend forwards"))] + +use std::{ + collections::{HashMap, HashSet}, + io, + net::{IpAddr, Ipv6Addr, SocketAddr}, + sync::{Arc, PoisonError, RwLock}, + task::{Context, Poll, ready}, +}; + +use hmac::{KeyInit, Mac}; +use moq_sock::udp::EcnCodepoint; + +type Hmac = hmac::Hmac; + +/// Where a QUIC-LB connection ID carries its server ID: after the byte encoding its length. +pub(crate) const ID_OFFSET: usize = 1; + +/// The forward header's version, so a mismatched peer's forwards are dropped rather than misread. +const VERSION: u8 = 0; +/// The truncated HMAC-SHA256 tag that authenticates a forward. +const TAG_LEN: usize = 16; +/// The flag marking a forward that knows the address its packet arrived on. +const ARRIVAL: u8 = 0b100; +/// The header ahead of a forwarded packet: version, flags (ECN and [`ARRIVAL`]), the client's +/// port and IP, the arrival IP, and the tag over all of it and the packet. +pub(crate) const OVERHEAD: usize = 1 + 1 + 2 + 16 + 16 + TAG_LEN; + +/// Generates QUIC-LB connection IDs: a byte encoding the length, the server ID, then a random nonce. +pub(crate) struct ServerIdGenerator { + id: Box<[u8]>, + nonce_len: usize, +} + +impl ServerIdGenerator { + /// Name `id` in every connection ID, ahead of `nonce_len` random bytes. + pub(crate) fn new(id: &[u8], nonce_len: usize) -> Self { + Self { + id: id.into(), + nonce_len, + } + } +} + +impl moq_quic::ConnectionIdGenerator for ServerIdGenerator { + fn generate_cid(&mut self) -> moq_quic::ConnectionId { + use rand::RngExt; + let cid_len = self.cid_len(); + let mut cid = Vec::with_capacity(cid_len); + // The "self-encoded length" of server ID and nonce, which puts the server ID at `ID_OFFSET`. + cid.push((cid_len - 1) as u8); + cid.extend(self.id.iter()); + cid.extend(rand::rng().random_iter::().take(self.nonce_len)); + moq_quic::ConnectionId::new(cid.as_slice()) + } + + fn cid_len(&self) -> usize { + ID_OFFSET + self.id.len() + self.nonce_len + } + + fn cid_lifetime(&self) -> Option { + None + } +} + +/// The other servers behind the shared address, by server ID, replaced whole on each update. +#[derive(Clone, Debug, Default)] +pub(crate) struct Peers(Arc>); + +/// A snapshot of [`Peers`]. +#[derive(Debug, Default)] +pub(crate) struct Map { + by_id: HashMap, SocketAddr>, + /// The forward sockets accepted as sources. + addrs: HashSet, +} + +impl Map { + /// Name `id`'s forward socket. + pub(crate) fn insert(&mut self, id: &[u8], addr: SocketAddr) { + self.by_id.insert(id.into(), addr); + self.addrs.insert(addr); + } + + /// Every peer, by server ID. + #[cfg(test)] + pub(crate) fn peers(&self) -> &HashMap, SocketAddr> { + &self.by_id + } +} + +impl Peers { + /// Replace every peer at once. + pub(crate) fn set(&self, map: Map) { + *self.0.write().unwrap_or_else(PoisonError::into_inner) = map; + } + + fn get(&self, id: &[u8]) -> Option { + self.0 + .read() + .unwrap_or_else(PoisonError::into_inner) + .by_id + .get(id) + .copied() + } + + fn contains(&self, addr: &SocketAddr) -> bool { + self.0.read().unwrap_or_else(PoisonError::into_inner).addrs.contains(addr) + } +} + +/// A forwarded packet, as the client sent it. +pub(crate) struct Unwrapped<'a> { + pub remote: SocketAddr, + pub local_ip: Option, + pub ecn: Option, + pub packet: &'a [u8], +} + +/// One server's forwarding: its server ID, its peers, and the socket forwards travel on. +#[derive(Debug)] +pub(crate) struct Forward { + socket: tokio::net::UdpSocket, + /// The same socket, sending without waiting on the reactor to report it writable, which a + /// fresh socket is not yet. + send: std::net::UdpSocket, + id: Box<[u8]>, + peers: Peers, + key: Hmac, + /// Scratch for one forward, sent or received. + buf: Box<[u8]>, +} + +impl Forward { + /// Forward on `socket` between the servers in `peers`, this one named `id` in its connection IDs. + /// + /// `max_udp_payload_size` bounds the packets a client sends, so a forward never exceeds it + /// plus [`OVERHEAD`]. The socket may fragment: a full-size packet plus the header can exceed + /// the path between servers, which carries it anyway. + pub(crate) fn new( + socket: std::net::UdpSocket, + id: &[u8], + peers: Peers, + key: &[u8; 32], + max_udp_payload_size: u16, + ) -> io::Result { + socket.set_nonblocking(true)?; + let send = socket.try_clone()?; + let socket = { + let _runtime = crate::runtime::enter(); + tokio::net::UdpSocket::from_std(socket)? + }; + Ok(Self { + socket, + send, + id: id.into(), + peers, + key: Hmac::new_from_slice(key).expect("HMAC accepts any key length"), + buf: vec![0; usize::from(max_udp_payload_size) + OVERHEAD].into(), + }) + } + + /// Forward `packet` if its connection ID names another server, returning whether it was taken. + /// + /// A packet naming a server missing from the peers is taken and dropped: answering it here + /// would reset a connection another server owns. + pub(crate) fn route( + &mut self, + packet: &[u8], + remote: SocketAddr, + local_ip: Option, + ecn: Option, + ) -> bool { + let Some(id) = self.foreign(packet) else { + return false; + }; + let Some(peer) = self.peers.get(id) else { + tracing::trace!(id = %hex::encode(id), "dropping a packet for an unknown server"); + return true; + }; + // A client never sends more than the advertised payload size, so a longer datagram is + // not a packet for one of our connections. + let Some(len) = self.seal(packet, remote, local_ip, ecn) else { + return true; + }; + match self.send.send_to(&self.buf[..len], peer) { + Ok(_) => {} + // Lost like any datagram on a full socket; the client retransmits. + Err(err) if err.kind() == io::ErrorKind::WouldBlock => {} + Err(err) => tracing::debug!(%err, %peer, "failed to forward a packet"), + } + true + } + + /// Wrap `packet` into the scratch buffer, returning the forward's length, unless it would not fit. + fn seal( + &mut self, + packet: &[u8], + remote: SocketAddr, + local_ip: Option, + ecn: Option, + ) -> Option { + let buf = self.buf.get_mut(..OVERHEAD + packet.len())?; + buf[0] = VERSION; + buf[1] = ecn.map_or(0, |ecn| ecn as u8) | local_ip.map_or(0, |_| ARRIVAL); + buf[2..4].copy_from_slice(&remote.port().to_be_bytes()); + buf[4..20].copy_from_slice(&ipv6(remote.ip()).octets()); + buf[20..36].copy_from_slice(&local_ip.map_or(Ipv6Addr::UNSPECIFIED, ipv6).octets()); + buf[OVERHEAD..].copy_from_slice(packet); + let tag = tag(&self.key, buf).finalize().into_bytes(); + buf[OVERHEAD - TAG_LEN..OVERHEAD].copy_from_slice(&tag[..TAG_LEN]); + Some(buf.len()) + } + + /// Receive the next authentic forward, or `None` for a datagram that is not one. + pub(crate) fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll>>> { + let mut read = tokio::io::ReadBuf::new(&mut self.buf); + let from = ready!(self.socket.poll_recv_from(cx, &mut read))?; + let len = read.filled().len(); + Poll::Ready(Ok(self.open(from, len))) + } + + /// Authenticate the forward in the first `len` bytes of the scratch buffer. + fn open(&self, from: SocketAddr, len: usize) -> Option> { + // Checked before the tag, so a stranger costs a lookup and never a hash. + if !self.peers.contains(&from) { + return None; + } + let buf = &self.buf[..len]; + if len < OVERHEAD || buf[0] != VERSION { + return None; + } + tag(&self.key, buf) + .verify_truncated_left(&buf[OVERHEAD - TAG_LEN..OVERHEAD]) + .ok()?; + + let packet = &buf[OVERHEAD..]; + // Never forward again, and never answer for a connection another server owns: a peer + // with a stale view of who owns an ID would otherwise loop or reset it. + if self.foreign(packet).is_some() || packet.first().is_none_or(|first| first & 0x80 != 0) { + return None; + } + + let flags = buf[1]; + let port = u16::from_be_bytes([buf[2], buf[3]]); + let ip = |at: usize| canonical(Ipv6Addr::from(<[u8; 16]>::try_from(&buf[at..at + 16]).unwrap())); + Some(Unwrapped { + remote: SocketAddr::new(ip(4), port), + local_ip: (flags & ARRIVAL != 0).then(|| ip(20)), + ecn: EcnCodepoint::from_bits(flags), + packet, + }) + } + + /// The server ID in `packet`'s connection ID, when it is a short header naming another server. + fn foreign<'a>(&self, packet: &'a [u8]) -> Option<&'a [u8]> { + if packet.first()? & 0x80 != 0 { + return None; + } + let start = 1 + ID_OFFSET; + let id = packet.get(start..start + self.id.len())?; + (id != &*self.id).then_some(id) + } +} + +/// The MAC over `forward`'s header, short of the tag itself, and its packet. +fn tag(key: &Hmac, forward: &[u8]) -> Hmac { + let mut mac = key.clone(); + mac.update(&forward[..OVERHEAD - TAG_LEN]); + mac.update(&forward[OVERHEAD..]); + mac +} + +fn ipv6(ip: IpAddr) -> Ipv6Addr { + match ip { + IpAddr::V4(ip) => ip.to_ipv6_mapped(), + IpAddr::V6(ip) => ip, + } +} + +fn canonical(ip: Ipv6Addr) -> IpAddr { + match ip.to_ipv4_mapped() { + Some(ip) => IpAddr::V4(ip), + None => IpAddr::V6(ip), + } +} + +#[cfg(test)] +mod tests { + use std::{future::poll_fn, net::Ipv4Addr}; + + use super::*; + + const KEY: [u8; 32] = [7; 32]; + + type Opened = (SocketAddr, Option, Option, Vec); + + fn socket() -> std::net::UdpSocket { + std::net::UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap() + } + + /// The server named `id`, forwarding on `socket` under `key`. + fn server(id: u8, socket: std::net::UdpSocket, peers: &Peers, key: &[u8; 32]) -> Forward { + Forward::new(socket, &[id], peers.clone(), key, 1200).unwrap() + } + + /// A short header whose connection ID names server `id`. + fn short(id: u8) -> Vec { + vec![0x40, 5, id, 0xa, 0xb, 0xc, 0xd, 0x01, 0x02] + } + + /// The next datagram `forward` receives, opened. + async fn recv(forward: &mut Forward) -> Option { + let next = poll_fn(|cx| { + forward + .poll_recv(cx) + .map_ok(|opened| opened.map(|o| (o.remote, o.local_ip, o.ecn, o.packet.to_vec()))) + }); + tokio::time::timeout(std::time::Duration::from_secs(5), next) + .await + .expect("a datagram") + .unwrap() + } + + /// Servers 1 and 2, each listing both. + fn pair() -> (Forward, Forward, Peers) { + let (a, b) = (socket(), socket()); + let peers = Peers::default(); + let mut map = Map::default(); + map.insert(&[1], a.local_addr().unwrap()); + map.insert(&[2], b.local_addr().unwrap()); + peers.set(map); + (server(1, a, &peers, &KEY), server(2, b, &peers, &KEY), peers) + } + + #[tokio::test] + async fn a_forward_arrives_as_the_client_sent_it() { + let (mut a, mut b, _) = pair(); + let client: SocketAddr = "192.0.2.7:5000".parse().unwrap(); + let arrival: IpAddr = "198.51.100.1".parse().unwrap(); + + assert!(a.route(&short(2), client, Some(arrival), Some(EcnCodepoint::Ce))); + assert_eq!( + recv(&mut b).await, + Some((client, Some(arrival), Some(EcnCodepoint::Ce), short(2))) + ); + + // An IPv6 client, through a server that never learned the arrival address. + let client: SocketAddr = "[2001:db8::7]:5000".parse().unwrap(); + assert!(a.route(&short(2), client, None, None)); + assert_eq!(recv(&mut b).await, Some((client, None, None, short(2)))); + } + + /// What stays: this server's own connection IDs, long headers, and anything too short to + /// carry a server ID. A short header for a server nobody listed is taken and dropped. + #[tokio::test] + async fn only_another_servers_short_headers_leave() { + let (mut a, _b, _) = pair(); + let client: SocketAddr = "192.0.2.7:5000".parse().unwrap(); + assert!(!a.route(&short(1), client, None, None)); + let mut long = short(2); + long[0] = 0xc0; + assert!(!a.route(&long, client, None, None)); + assert!(!a.route(&[0x40, 5], client, None, None)); + assert!(a.route(&short(3), client, None, None)); + } + + /// A forward opens only from a listed peer, under the shared key, and only for this server: a + /// peer with a stale map must not make it forward again or answer for another server. + #[tokio::test] + async fn a_forward_needs_a_peer_the_key_and_this_server() { + let (mut a, mut b, peers) = pair(); + let client: SocketAddr = "192.0.2.7:5000".parse().unwrap(); + let b_addr = b.socket.local_addr().unwrap(); + + // Not listed, with the right key. + let mut stranger = server(1, socket(), &peers, &KEY); + assert!(stranger.route(&short(2), client, None, None)); + assert_eq!(recv(&mut b).await, None); + + // Listed, with the wrong key. + let mut forged = server(3, socket(), &peers, &[8; 32]); + let mut map = Map::default(); + map.insert(&[1], a.socket.local_addr().unwrap()); + map.insert(&[2], b_addr); + map.insert(&[3], forged.socket.local_addr().unwrap()); + peers.set(map); + assert!(forged.route(&short(2), client, None, None)); + assert_eq!(recv(&mut b).await, None); + + // Server 1 sending server 2 a packet of its own, as a stale map would. + let len = a.seal(&short(1), client, None, None).unwrap(); + a.send.send_to(&a.buf[..len], b_addr).unwrap(); + assert_eq!(recv(&mut b).await, None); + + // Still open to a real forward. + assert!(a.route(&short(2), client, None, None)); + assert!(recv(&mut b).await.is_some()); + } +} diff --git a/rs/moq-tokio/src/quic/driver/mod.rs b/rs/moq-tokio/src/quic/driver/mod.rs index e76e92a18b0..9fad9b504d5 100644 --- a/rs/moq-tokio/src/quic/driver/mod.rs +++ b/rs/moq-tokio/src/quic/driver/mod.rs @@ -47,6 +47,7 @@ pub mod connection; pub mod endpoint; +pub(crate) mod forward; mod incoming; mod mutex; pub mod recv; diff --git a/rs/moq-tokio/src/quic/driver/tests.rs b/rs/moq-tokio/src/quic/driver/tests.rs index a595f16b4e9..ed7554899b6 100755 --- a/rs/moq-tokio/src/quic/driver/tests.rs +++ b/rs/moq-tokio/src/quic/driver/tests.rs @@ -304,11 +304,16 @@ impl EndpointFactory { self.endpoint_with_config(transport::Config::default()) } - fn endpoint_with_config(&self, transport_config: transport::Config) -> Endpoint { + fn server_config(&self, transport_config: transport::Config) -> server::Config { let key = PrivateKeyDer::Pkcs8(self.cert.signing_key.serialize_der().into()); let mut server_config = crate::quic::driver::server::Config::with_single_cert(vec![self.cert.cert.der().clone()], key).unwrap(); - server_config.transport = transport_config.clone(); + server_config.transport = transport_config; + server_config + } + + fn endpoint_with_config(&self, transport_config: transport::Config) -> Endpoint { + let server_config = self.server_config(transport_config.clone()); let mut roots = RootCertStore::empty(); roots.add(self.cert.cert.der().clone()).unwrap(); @@ -1346,6 +1351,103 @@ async fn remote_address_follows_migration() { client_conn.close(0u32.into(), "done"); } +/// Three servers sit behind one front address. A connection whose handshake reached the first +/// keeps working while the front sends its packets to the second, then the third: each forwards +/// them to the first, which replies to the client directly. +#[tokio::test] +async fn forwarded_packets_reach_their_owner() { + use crate::quic::driver::forward::{Forward, Map, Peers, ServerIdGenerator}; + + let _guard = subscribe(); + let localhost = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); + let factory = EndpointFactory::new(); + let key = [7; 32]; + + let sockets: Vec = (0..3).map(|_| UdpSocket::bind(localhost).unwrap()).collect(); + let peers = Peers::default(); + let mut map = Map::default(); + for (id, socket) in (1u8..).zip(&sockets) { + map.insert(&[id], socket.local_addr().unwrap()); + } + peers.set(map); + let servers: Vec = (1u8..) + .zip(sockets) + .map(|(id, socket)| { + let mut config = factory.endpoint_config.clone(); + config.cid_generator = Some(Arc::new(move || Box::new(ServerIdGenerator::new(&[id], 4)))); + let forward = Forward::new(socket, &[id], peers.clone(), &key, config.max_udp_payload_size).unwrap(); + Endpoint::forwarding( + config.with_server(factory.server_config(transport::Config::default())), + UdpSocket::bind(localhost).unwrap(), + forward, + ) + .unwrap() + }) + .collect(); + + // The front end: the client's packets go to whichever server it names, from one address, and + // a reply from any server goes back to the client. + let front = tokio::net::UdpSocket::bind(localhost).await.unwrap(); + let front_addr = front.local_addr().unwrap(); + let back = tokio::net::UdpSocket::bind(localhost).await.unwrap(); + let back_addr = back.local_addr().unwrap(); + let target = Arc::new(Mutex::new(servers[0].local_addr())); + let front_task = tokio::spawn({ + let target = target.clone(); + async move { + let (mut up, mut down) = ([0; 1500], [0; 1500]); + let mut client = None; + loop { + tokio::select! { + received = front.recv_from(&mut up) => { + let (len, from) = received.unwrap(); + client = Some(from); + let to = *target.lock().unwrap(); + back.send_to(&up[..len], to).await.unwrap(); + } + received = back.recv_from(&mut down) => { + let (len, _) = received.unwrap(); + front.send_to(&down[..len], client.unwrap()).await.unwrap(); + } + } + } + } + }); + + let owner = servers[0].clone(); + let server_task = tokio::spawn(async move { + let conn = owner.accept().await.unwrap().accept().unwrap().await.unwrap(); + loop { + let Ok(stream) = conn.accept_bi().await else { break }; + // The client's packets keep arriving from the address its handshake took, whichever + // server they reached first. + assert_eq!(conn.remote_address(), back_addr); + echo(stream).await; + } + }); + + let client = factory.endpoint(); + let conn = client.dial(front_addr, "localhost").unwrap().await.unwrap(); + for (round, server) in [0, 1, 2, 0].into_iter().enumerate() { + *target.lock().unwrap() = servers[server].local_addr(); + let message = format!("round {round} through server {server}"); + let (mut send, mut recv) = conn.open_bi(Default::default()).await.unwrap(); + send.write_all(message.as_bytes()).await.unwrap(); + let _ = send.finish(); + let echoed = tokio::time::timeout(Duration::from_secs(10), recv.read_to_end(1024)) + .await + .expect("the owner answered") + .unwrap(); + assert_eq!(echoed, message.as_bytes()); + } + assert_eq!(servers[1].open_connections(), 0); + assert_eq!(servers[2].open_connections(), 0); + + conn.close(0u32.into(), "done"); + server_task.await.unwrap(); + front_task.abort(); +} + /// A send group from another connection panics where it is passed in, without poisoning the /// connection's lock: a poisoned lock panics again in every handle's drop, aborting the process. #[tokio::test] From 953d523a3af7c0e296ce1ea85f14a0b3dcb73f84 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 02:17:11 -0800 Subject: [PATCH 3/8] style: format forwarding changes Co-Authored-By: Claude Opus 5.5 --- rs/moq-relay/src/uring.rs | 13 +++++++------ rs/moq-tokio/src/listen.rs | 5 ++++- rs/moq-tokio/src/quic/backend.rs | 9 +++++++-- rs/moq-tokio/src/quic/driver/forward.rs | 6 +++++- 4 files changed, 23 insertions(+), 10 deletions(-) diff --git a/rs/moq-relay/src/uring.rs b/rs/moq-relay/src/uring.rs index a1e44dd409f..96cd539073f 100644 --- a/rs/moq-relay/src/uring.rs +++ b/rs/moq-relay/src/uring.rs @@ -147,12 +147,13 @@ impl Workers { tls: listen_tls, preferred_v4, preferred_v6, - lb: moq_tokio::listen::LoadBalancer { - id: lb_id, - nonce: _, - // Needs an id, refused below. - forward: _, - }, + lb: + moq_tokio::listen::LoadBalancer { + id: lb_id, + nonce: _, + // Needs an id, refused below. + forward: _, + }, reset_secret, } = server.listen.resolve()?; let quic = &server.quic; diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs index 9d3da04f79b..c2b4dc2e5bd 100644 --- a/rs/moq-tokio/src/listen.rs +++ b/rs/moq-tokio/src/listen.rs @@ -814,7 +814,10 @@ peers = "peers.toml" complete.replace("bind = \"10.0.0.1:4443\"\n", ""), "requires --listen-quic-lb-forward-bind", ), - (complete.replace("key = \"k\"\n", ""), "requires --listen-quic-lb-forward-key"), + ( + complete.replace("key = \"k\"\n", ""), + "requires --listen-quic-lb-forward-key", + ), ( complete.replace("peers = \"p\"\n", ""), "requires --listen-quic-lb-forward-peers", diff --git a/rs/moq-tokio/src/quic/backend.rs b/rs/moq-tokio/src/quic/backend.rs index 7dec3ac467e..b04746fb390 100644 --- a/rs/moq-tokio/src/quic/backend.rs +++ b/rs/moq-tokio/src/quic/backend.rs @@ -905,10 +905,15 @@ fn read_peers(path: &std::path::Path, own: &ServerId, bind: net::SocketAddr) -> let mut map = driver::forward::Map::default(); for (id, addr) in peers { if id.len() != own.len() { - return Err(invalid(format!("{id:?} is not {} bytes like this server's ID", own.len()))); + return Err(invalid(format!( + "{id:?} is not {} bytes like this server's ID", + own.len() + ))); } if addr.ip().is_unspecified() || addr.is_ipv4() != bind.is_ipv4() { - return Err(invalid(format!("{id:?} has an address {bind} cannot forward to: {addr}"))); + return Err(invalid(format!( + "{id:?} has an address {bind} cannot forward to: {addr}" + ))); } if id != *own { map.insert(&id.0, addr); diff --git a/rs/moq-tokio/src/quic/driver/forward.rs b/rs/moq-tokio/src/quic/driver/forward.rs index f172af6bfce..78e7736f420 100644 --- a/rs/moq-tokio/src/quic/driver/forward.rs +++ b/rs/moq-tokio/src/quic/driver/forward.rs @@ -115,7 +115,11 @@ impl Peers { } fn contains(&self, addr: &SocketAddr) -> bool { - self.0.read().unwrap_or_else(PoisonError::into_inner).addrs.contains(addr) + self.0 + .read() + .unwrap_or_else(PoisonError::into_inner) + .addrs + .contains(addr) } } From e2a5c200b89496c6a351d838f5513d02fa80cb5e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 02:22:16 -0800 Subject: [PATCH 4/8] fix: register the forward settings and use the ambient clock in tests Co-Authored-By: Claude Opus 5.5 --- rs/moq-tokio/src/quic/driver/forward.rs | 2 +- rs/moq-tokio/src/quic/driver/tests.rs | 2 +- rs/moq-tokio/src/settings.rs | 17 +++++++++++++++++ 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/rs/moq-tokio/src/quic/driver/forward.rs b/rs/moq-tokio/src/quic/driver/forward.rs index 78e7736f420..5771ce5729f 100644 --- a/rs/moq-tokio/src/quic/driver/forward.rs +++ b/rs/moq-tokio/src/quic/driver/forward.rs @@ -330,7 +330,7 @@ mod tests { .poll_recv(cx) .map_ok(|opened| opened.map(|o| (o.remote, o.local_ip, o.ecn, o.packet.to_vec()))) }); - tokio::time::timeout(std::time::Duration::from_secs(5), next) + kio::clock::timeout(std::time::Duration::from_secs(5), next) .await .expect("a datagram") .unwrap() diff --git a/rs/moq-tokio/src/quic/driver/tests.rs b/rs/moq-tokio/src/quic/driver/tests.rs index ed7554899b6..e4f39fceb7d 100755 --- a/rs/moq-tokio/src/quic/driver/tests.rs +++ b/rs/moq-tokio/src/quic/driver/tests.rs @@ -1434,7 +1434,7 @@ async fn forwarded_packets_reach_their_owner() { let (mut send, mut recv) = conn.open_bi(Default::default()).await.unwrap(); send.write_all(message.as_bytes()).await.unwrap(); let _ = send.finish(); - let echoed = tokio::time::timeout(Duration::from_secs(10), recv.read_to_end(1024)) + let echoed = timeout(Duration::from_secs(10), recv.read_to_end(1024)) .await .expect("the owner answered") .unwrap(); diff --git a/rs/moq-tokio/src/settings.rs b/rs/moq-tokio/src/settings.rs index 3b2c565674f..4a51f388826 100644 --- a/rs/moq-tokio/src/settings.rs +++ b/rs/moq-tokio/src/settings.rs @@ -321,4 +321,21 @@ struct ListenLb { #[usage(env = "MOQ_LISTEN_QUIC_LB_NONCE", cli("--listen-quic-lb-nonce"))] nonce: Option, + + #[usage(flatten)] + forward: ListenLbForward, +} + +/// `[listen.lb.forward]` / `--listen-quic-lb-forward-*`. +#[derive(usage_rs::Config)] +#[usage(prefix = "listen.lb.forward")] +struct ListenLbForward { + #[usage(env = "MOQ_LISTEN_QUIC_LB_FORWARD_BIND", cli("--listen-quic-lb-forward-bind"))] + bind: Option, + + #[usage(env = "MOQ_LISTEN_QUIC_LB_FORWARD_KEY", cli("--listen-quic-lb-forward-key"))] + key: Option, + + #[usage(env = "MOQ_LISTEN_QUIC_LB_FORWARD_PEERS", cli("--listen-quic-lb-forward-peers"))] + peers: Option, } From 9a2574d0478499b520d9d125c73d71a2dec2a9a9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 14:07:23 -0800 Subject: [PATCH 5/8] fix(tokio): keep a local stateless reset from being forwarded A stateless reset carries a random connection ID, so its server-ID bytes usually name another server, and forwarding took it before the endpoint's reset-token lookup could. A peer's reset of one of this server's connections, such as a mesh relay that restarted, was forwarded or dropped, and the dead connection retransmitted until it timed out. moq-quic gains Endpoint::is_stateless_reset, and Forward::route checks it for a packet naming another server before forwarding it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-quic/src/endpoint.rs | 14 +++++++++++ rs/moq-quic/src/tests/mod.rs | 7 ++++++ rs/moq-tokio/src/quic/driver/endpoint.rs | 5 ++-- rs/moq-tokio/src/quic/driver/forward.rs | 32 +++++++++++++++--------- 4 files changed, 44 insertions(+), 14 deletions(-) diff --git a/rs/moq-quic/src/endpoint.rs b/rs/moq-quic/src/endpoint.rs index 4c417fac361..dcf994331ab 100644 --- a/rs/moq-quic/src/endpoint.rs +++ b/rs/moq-quic/src/endpoint.rs @@ -1027,6 +1027,20 @@ impl Endpoint { self.connections.len() } + /// Whether `datagram` from `remote` ends in the stateless reset token a connection's peer + /// issued, so [`handle`](Self::handle) would route it to that connection + /// + /// A stateless reset carries a random connection ID, so a caller that routes datagrams by + /// connection ID before handing them over checks this first. + pub fn is_stateless_reset(&self, remote: SocketAddr, datagram: &[u8]) -> bool { + datagram.len().checked_sub(RESET_TOKEN_SIZE).is_some_and(|start| { + self.index + .connection_reset_tokens + .get(remote, &datagram[start..]) + .is_some() + }) + } + /// Counter for the number of bytes currently used /// in the buffers for Initial messages for pending incoming connections /// and accepts that are still being finalized diff --git a/rs/moq-quic/src/tests/mod.rs b/rs/moq-quic/src/tests/mod.rs index 304be041a60..732f57d7f37 100644 --- a/rs/moq-quic/src/tests/mod.rs +++ b/rs/moq-quic/src/tests/mod.rs @@ -549,6 +549,13 @@ fn server_stateless_reset() { // Force the server to generate the smallest possible stateless reset pair.client.connections.get_mut(&client_ch).unwrap().ping(); info!("resetting"); + pair.drive_client(); + pair.drive_server(); + // The client recognizes the reset before handling it, and not just any short header. + let (_, _, reset) = pair.client.inbound.back().unwrap(); + assert!(pair.client.endpoint.is_stateless_reset(pair.server.addr, reset)); + assert!(!pair.client.endpoint.is_stateless_reset(pair.server.addr, &[0x40; 64])); + assert!(!pair.client.endpoint.is_stateless_reset(pair.client.addr, reset)); pair.drive(); assert_matches!( pair.client_conn_mut(client_ch).poll(), diff --git a/rs/moq-tokio/src/quic/driver/endpoint.rs b/rs/moq-tokio/src/quic/driver/endpoint.rs index f2b4ca3d6d5..a86e723f6c6 100644 --- a/rs/moq-tokio/src/quic/driver/endpoint.rs +++ b/rs/moq-tokio/src/quic/driver/endpoint.rs @@ -945,8 +945,9 @@ impl RecvState { while !data.is_empty() { let buf = data.split_to(meta.stride.min(data.len())); if let Some(forward) = forward.as_deref_mut() - && forward.route(&buf, meta.addr, meta.dst_ip, meta.ecn) - { + && forward.route(&buf, meta.addr, meta.dst_ip, meta.ecn, || { + endpoint.is_stateless_reset(meta.addr, &buf) + }) { continue; } let received = Received { diff --git a/rs/moq-tokio/src/quic/driver/forward.rs b/rs/moq-tokio/src/quic/driver/forward.rs index 5771ce5729f..55dce4bcaec 100644 --- a/rs/moq-tokio/src/quic/driver/forward.rs +++ b/rs/moq-tokio/src/quic/driver/forward.rs @@ -177,17 +177,22 @@ impl Forward { /// Forward `packet` if its connection ID names another server, returning whether it was taken. /// /// A packet naming a server missing from the peers is taken and dropped: answering it here - /// would reset a connection another server owns. + /// would reset a connection another server owns. One that `is_local_reset` recognizes stays, + /// since a stateless reset's connection ID is random and may seem to name another server. pub(crate) fn route( &mut self, packet: &[u8], remote: SocketAddr, local_ip: Option, ecn: Option, + is_local_reset: impl FnOnce() -> bool, ) -> bool { let Some(id) = self.foreign(packet) else { return false; }; + if is_local_reset() { + return false; + } let Some(peer) = self.peers.get(id) else { tracing::trace!(id = %hex::encode(id), "dropping a packet for an unknown server"); return true; @@ -353,7 +358,7 @@ mod tests { let client: SocketAddr = "192.0.2.7:5000".parse().unwrap(); let arrival: IpAddr = "198.51.100.1".parse().unwrap(); - assert!(a.route(&short(2), client, Some(arrival), Some(EcnCodepoint::Ce))); + assert!(a.route(&short(2), client, Some(arrival), Some(EcnCodepoint::Ce), || false)); assert_eq!( recv(&mut b).await, Some((client, Some(arrival), Some(EcnCodepoint::Ce), short(2))) @@ -361,22 +366,25 @@ mod tests { // An IPv6 client, through a server that never learned the arrival address. let client: SocketAddr = "[2001:db8::7]:5000".parse().unwrap(); - assert!(a.route(&short(2), client, None, None)); + assert!(a.route(&short(2), client, None, None, || false)); assert_eq!(recv(&mut b).await, Some((client, None, None, short(2)))); } - /// What stays: this server's own connection IDs, long headers, and anything too short to - /// carry a server ID. A short header for a server nobody listed is taken and dropped. + /// What stays: this server's own connection IDs, long headers, anything too short to carry a + /// server ID, and a stateless reset for one of this server's connections, whose random + /// connection ID may seem to name another. A short header for a server nobody listed is taken + /// and dropped. #[tokio::test] async fn only_another_servers_short_headers_leave() { let (mut a, _b, _) = pair(); let client: SocketAddr = "192.0.2.7:5000".parse().unwrap(); - assert!(!a.route(&short(1), client, None, None)); + assert!(!a.route(&short(1), client, None, None, || false)); let mut long = short(2); long[0] = 0xc0; - assert!(!a.route(&long, client, None, None)); - assert!(!a.route(&[0x40, 5], client, None, None)); - assert!(a.route(&short(3), client, None, None)); + assert!(!a.route(&long, client, None, None, || false)); + assert!(!a.route(&[0x40, 5], client, None, None, || false)); + assert!(a.route(&short(3), client, None, None, || false)); + assert!(!a.route(&short(2), client, None, None, || true)); } /// A forward opens only from a listed peer, under the shared key, and only for this server: a @@ -389,7 +397,7 @@ mod tests { // Not listed, with the right key. let mut stranger = server(1, socket(), &peers, &KEY); - assert!(stranger.route(&short(2), client, None, None)); + assert!(stranger.route(&short(2), client, None, None, || false)); assert_eq!(recv(&mut b).await, None); // Listed, with the wrong key. @@ -399,7 +407,7 @@ mod tests { map.insert(&[2], b_addr); map.insert(&[3], forged.socket.local_addr().unwrap()); peers.set(map); - assert!(forged.route(&short(2), client, None, None)); + assert!(forged.route(&short(2), client, None, None, || false)); assert_eq!(recv(&mut b).await, None); // Server 1 sending server 2 a packet of its own, as a stale map would. @@ -408,7 +416,7 @@ mod tests { assert_eq!(recv(&mut b).await, None); // Still open to a real forward. - assert!(a.route(&short(2), client, None, None)); + assert!(a.route(&short(2), client, None, None, || false)); assert!(recv(&mut b).await.is_some()); } } From 19e5126bc39b1f1d2c2e91e0f672f284da746070 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 14:27:25 -0800 Subject: [PATCH 6/8] doc(tokio): migration note for the LoadBalancer forward field Co-Authored-By: Claude Opus 5.5 --- rs/moq-tokio/MIGRATE.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/rs/moq-tokio/MIGRATE.md b/rs/moq-tokio/MIGRATE.md index ffbe2a04c7f..1cabb27479f 100644 --- a/rs/moq-tokio/MIGRATE.md +++ b/rs/moq-tokio/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `listen::LoadBalancer` has a `forward` field ([#7539](https://github.com/moq-dev/moq/pull/7539)) + +`listen::LoadBalancer` gained `forward: listen::Forward`, the `[listen.lb.forward]` settings that forward QUIC packets to the server their connection ID names. +The struct is not `#[non_exhaustive]`, so add `..Default::default()` to a `LoadBalancer { id, nonce }` literal, which leaves forwarding off. +An exhaustive destructure, such as a QUIC runtime mapping the resolved listener settings, must name `forward`: honor it, or ignore it with `forward: _` only where `lb.id` is already refused. + ### `tls://` listeners verify client certificates ([#7358](https://github.com/moq-dev/moq/pull/7358)) A `listen.tcp.tls` listener serves the listen policy's client-certificate verifier, so `--listen-tls-root` and pinned `tls.peers` apply to it as they do to QUIC, and a stream-only server with that listener accepts them instead of returning `Error::MtlsUnsupported`. From 36eeef4f55e842d0d06e44b692e7182b52afac09 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 21:16:17 -0800 Subject: [PATCH 7/8] fix(tokio): address review on forwarding docs and the dead_code lint Move `unix.allow` back into its paragraph and tighten the forwarding section; allow rather than expect dead_code on `Endpoint::forwarding`, since the driver's tests call it without the `quic` feature. Co-Authored-By: Claude Opus 5.5 --- doc/bin/relay/config.md | 32 +++++++++++------------- rs/moq-tokio/src/quic/driver/endpoint.rs | 2 +- 2 files changed, 15 insertions(+), 19 deletions(-) diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md index 27fa228bf9b..441352ce03b 100644 --- a/doc/bin/relay/config.md +++ b/doc/bin/relay/config.md @@ -87,32 +87,28 @@ The io\_uring runtime refuses `preferred_v4`, `preferred_v6`, and `lb.id`; use the tokio runtime when the listener needs these settings. `[listen.lb] nonce` needs an `id` in the same table; unset nonce uses 8. The `--listen-quic-lb-id` and `--listen-quic-lb-nonce` flags update that table. +`unix.allow` needs `unix.bind`. `[listen.lb.forward]` keeps a connection alive when its packets land on another relay behind -the same anycast address or load-balanced VIP, after a route change or a rebalance. That relay -reads the server ID in the packet's connection ID and forwards the packet to the owner's -forward socket, which processes it as if the client had sent it there. The owner keeps replying -straight to the client from the shared address its handshake arrived on, so only the -client-to-server direction takes the extra hop. A browser never leaves that address: Chrome's -WebTransport ignores the QUIC `preferred_address`. Handshakes are not forwarded, so the front -end must keep each new flow on one relay until it completes; a split handshake fails and the -client retries. - -The peers file maps every relay's server ID to its forward `bind`, and may list this relay too, -so one file can serve the whole cluster: +the same anycast address or VIP, after a route change or a rebalance. That relay reads the +server ID in the connection ID and forwards the packet to the owner's forward socket, which +handles it as if the client had sent it there and keeps replying from the shared address. +Handshakes are not forwarded, so the front end must keep each new flow on one relay until it +completes. + +The peers file maps each relay's server ID to its forward `bind`, and may list this relay too, +so one file serves the whole cluster: ```toml "01" = "10.0.0.1:4443" "02" = "10.0.0.2:4443" ``` -A packet for a server ID missing from it is dropped. A forward is accepted only from a listed -address and under the shared `key` (`openssl rand -hex 32 > forward.key`), since it carries a -client address the owner takes on trust. A changed file replaces the map; an invalid one is -logged and the last valid map kept, while an invalid one at startup stops it. Forwarding needs -`lb.id` and all three settings, and the -`--listen-quic-lb-forward-bind`, `-key`, and `-peers` flags update that table. -`unix.allow` needs `unix.bind`. +A packet for an unlisted server ID is dropped. A forward is accepted only from a listed address +and under the shared `key` (`openssl rand -hex 32 > forward.key`). A changed file replaces the +map; an invalid one is logged and the last valid map kept, or stops startup. Forwarding needs +`lb.id` and all three settings, which the `--listen-quic-lb-forward-bind`, `-key`, and `-peers` +flags update. ## \[quic] diff --git a/rs/moq-tokio/src/quic/driver/endpoint.rs b/rs/moq-tokio/src/quic/driver/endpoint.rs index 663845ec5ff..6d8af4c1b16 100644 --- a/rs/moq-tokio/src/quic/driver/endpoint.rs +++ b/rs/moq-tokio/src/quic/driver/endpoint.rs @@ -106,7 +106,7 @@ impl Endpoint { /// [`new`](Self::new), forwarding each packet whose QUIC-LB server ID names another server /// and accepting the packets those servers forward here. - #[cfg_attr(not(feature = "quic"), expect(dead_code, reason = "only the QUIC backend forwards"))] + #[cfg_attr(not(feature = "quic"), allow(dead_code, reason = "only the QUIC backend forwards"))] pub(crate) fn forwarding(config: Config, socket: impl Into, forward: Forward) -> Result { Self::start(config, socket.into(), Some(forward)) } From 33abe453bd35032a4a4d34c8c0aae6498484cbbb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 22:56:52 -0800 Subject: [PATCH 8/8] fix(tokio): close releases the QUIC-LB forward socket The forward socket outlived Endpoint::close while connections drained, so a listener restarted on the same configuration failed to bind it. Drop it with the main socket on close and when the driver ends. Co-Authored-By: Claude Opus 5.5 --- rs/moq-tokio/src/quic/driver/endpoint.rs | 6 ++++- rs/moq-tokio/src/quic/driver/tests.rs | 32 ++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/rs/moq-tokio/src/quic/driver/endpoint.rs b/rs/moq-tokio/src/quic/driver/endpoint.rs index 6d8af4c1b16..bf9cf4fceea 100644 --- a/rs/moq-tokio/src/quic/driver/endpoint.rs +++ b/rs/moq-tokio/src/quic/driver/endpoint.rs @@ -243,8 +243,11 @@ impl Endpoint { async move { endpoint.wait_all_draining().await; endpoint.inner.slot.release(); + let state = &mut *endpoint.inner.state.lock().unwrap(); // A rebind keeps the old socket until traffic arrives on the new one. - endpoint.inner.state.lock().unwrap().prev_socket = None; + state.prev_socket = None; + // Nothing reaches the endpoint to forward once its socket is gone. + state.forward = None; } } @@ -394,6 +397,7 @@ impl Drop for EndpointDriver { self.0.slot.release(); let endpoint = &mut *self.0.state.lock().unwrap(); endpoint.prev_socket = None; + endpoint.forward = None; endpoint.driver_lost = true; self.0.shared.incoming.notify_waiters(); // Drop all outgoing channels, signaling the termination of the endpoint to the associated diff --git a/rs/moq-tokio/src/quic/driver/tests.rs b/rs/moq-tokio/src/quic/driver/tests.rs index f310cc753a4..64da9a12878 100755 --- a/rs/moq-tokio/src/quic/driver/tests.rs +++ b/rs/moq-tokio/src/quic/driver/tests.rs @@ -1474,6 +1474,38 @@ async fn forwarded_packets_reach_their_owner() { front_task.abort(); } +/// Closing a forwarding endpoint frees its forward port with its main one, even while a handle +/// keeps the endpoint alive, so a listener restarted on the same configuration can bind both. +#[tokio::test] +async fn close_releases_the_forward_socket() { + use crate::quic::driver::forward::{Forward, Peers}; + + let _guard = subscribe(); + let localhost = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); + let factory = EndpointFactory::new(); + let forward_socket = UdpSocket::bind(localhost).unwrap(); + let forward_addr = forward_socket.local_addr().unwrap(); + let config = factory.endpoint_config.clone(); + let forward = Forward::new( + forward_socket, + &[1], + Peers::default(), + &[7; 32], + config.max_udp_payload_size, + ) + .unwrap(); + let endpoint = Endpoint::forwarding( + config.with_server(factory.server_config(transport::Config::default())), + UdpSocket::bind(localhost).unwrap(), + forward, + ) + .unwrap(); + + endpoint.close(0u32.into(), "done").await; + UdpSocket::bind(forward_addr).expect("the forward port is free once close returns"); + drop(endpoint); +} + /// A send group from another connection panics where it is passed in, without poisoning the /// connection's lock: a poisoned lock panics again in every handle's drop, aborting the process. #[tokio::test]