feat(tokio)!: forward QUIC packets to their owner by QUIC-LB server ID

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"
@@ -80,6 +89,27 @@ use the tokio runtime when the listener needs these settings.
 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 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 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]
 
 Transport tuning, applied to accepted and dialed connections alike.

quest/m2/README.md

@@ -33,7 +33,6 @@ lite-07; bitrate enforcement is not part of this release.
 - [One Ruff pin and Swift formatting on aarch64-linux](/quest/m2/format-gate-followups.md) - only the dev shell pins Ruff, and aarch64-linux checks Swift formatting once nixpkgs builds `swiftc` there
 - [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
-- [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

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

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

@@ -544,6 +544,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(),

rs/moq-relay/src/uring.rs

@@ -147,7 +147,13 @@ 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;

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,

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`.

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<crate::quic::LoadBalancer> {
+		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,46 @@ 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 +981,66 @@ pub struct LoadBalancer {
 	)]
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub nonce: Option<usize>,
+
+	/// 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<std::net::SocketAddr>,
+
+	/// 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<std::path::PathBuf>,
+
+	/// 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<std::path::PathBuf>,
+}
+
+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()
+	}
 }