fix(relay): backport the mesh gossip fixes and relay counters to release

doc/bin/relay/cluster.md

@@ -70,6 +70,26 @@ For a full mesh, list every other relay, or serve the list from `connect_api`.
 A session carries both directions, so one dial per pair is enough; listing a
 pair on both sides opens a redundant second session.
 
+`cluster.max_hops` bounds the hop path this relay accepts or forwards, from 1
+to 32, with 32 as the default. Set it near the mesh's diameter after any single
+link or node failure, plus a margin, to cut off long detours during withdrawal
+path hunting. Count the publisher's hop too, and the random first hop a relay
+puts in front of a publisher that names none. A cap below that diameter can
+partition the mesh; a lower cap only drops longer paths and preserves route
+selection among the remaining paths. A peer hears the best route with room
+for this relay's hop, even when a cheaper route at the cap serves locally.
+
+```toml
+[cluster]
+max_hops = 6
+```
+
+The same setting is `--cluster-max-hops` or `MOQ_CLUSTER_MAX_HOPS`. It covers
+moq-lite-03 through moq-lite-07-wip and moq-transport drafts 17 through 22 when
+the MoQ Cluster extension is negotiated. A forwarded path counts this relay's
+own hop even when its peer appends it on receipt. Older versions and peers
+without hop paths hide their depth, which this setting cannot bound.
+
 ## Link costs
 
 Add `?cost=N` to a peer URL to route by price instead of hop count. An unpriced
@@ -119,7 +139,10 @@ pushing to it, so each end declares its own and the two need not match. Prices
 aren't static either. A publisher can re-price a live announcement, which is how
 a standby transcoder pool seeds a high cost and drops it once it's working, and
 a relay receiving a GOAWAY re-prices every route learned from that peer to the
-maximum so new subscriptions go elsewhere while existing ones finish.
+maximum so new subscriptions go elsewhere while existing ones finish. A
+relay keeps its session to a peer draining to restart until that peer closes
+it, and redials only then, so it reaches the replacement rather than the
+draining process.
 
 moq-lite-06 announcements carry two prices, *warm* and *cold*. Both accumulate
 identically today, so routing runs on link costs alone; the split reserves room
@@ -216,8 +239,8 @@ your own that grants the cluster CA) behind `--auth-url`. A relay on
 send no `cluster.token`: public rules refuse a token. LAN peers
 authenticate with the mDNS credential on `/.cluster/<credential>`, a secret
 the relay minted for itself and checks locally, and never receive
-`cluster.token`. Dials retry forever with capped backoff, so a rejected peer
-is loud in the logs rather than fatal. See [Authentication](/bin/relay/auth#mtls).
+`cluster.token`. Dials retry forever with jittered backoff capped at 30s, so a
+rejected peer is loud in the logs rather than fatal. See [Authentication](/bin/relay/auth#mtls).
 
 A relay records whether each route entered here or came from a peer, which the
 hop list alone cannot say: a client and a peer each add one hop. Routes over a

doc/bin/relay/config.md

@@ -156,6 +156,7 @@ node = "https://us-west.example.com/"                 # This relay's own URL.
 connect_api = "https://api.example.com/peers"        # Or fetch the peer list (JSON array of URLs and/or objects) live.
 token = "cluster.jwt"                                 # JWT for dials without an inline ?jwt=.
 id = 12345                                            # Stable Hop ID across restarts.
+max_hops = 32                                         # Longest accepted/forwarded hop path (1 to 32).
 
 [cluster.lan]                                         # Find peers on the LAN over mDNS.
 enabled = true

doc/bin/relay/http.md

@@ -47,8 +47,8 @@ listen = "127.0.0.1:9101"
 
 ### GET /metrics
 
-Prometheus text: bytes, frames, groups, subscriptions, viewers, and sessions
-split by `tier` and `role`, plus accept-loop counters per TCP listener. Alert
+Prometheus text: bytes, frames, groups, announces, subscriptions, viewers, and
+sessions split by `tier` and `role`, plus accept-loop counters per TCP listener. Alert
 on `moq_relay_accept_failures_total{class="exhausted"}`, which means the
 process ran out of a resource `accept` needs. Content dropped for drifting past
 a subscriber's budget is counted separately as `moq_relay_stale_bytes_total`
@@ -60,6 +60,12 @@ Traffic and session counters accumulate for the node's lifetime, including
 broadcasts and sessions that have ended. The stats publishing prefix (normally
 `.stats`) is excluded to avoid counting the feed's own traffic.
 
+`moq_relay_control_tx_bytes_total` and `moq_relay_control_rx_bytes_total`
+count control-message bytes by `tier` and `class` (`session`, `announce`, `track_info`, `subscribe`, `fetch`,
+`probe`, `auth`, `goaway`, `other`) to spot control-heavy sessions. They
+include message size prefixes but leave out media payload and transport
+overhead such as QUIC framing, ACKs, and retransmissions.
+
 With `--runtime-io-uring`, each QUIC worker thread also reports its own
 `moq_relay_uring_*` counters under a `worker` label: datagrams and syscalls
 (the ratios are the GRO/GSO batching and the syscall amortization the runtime

doc/concept/moq-lite.md

@@ -93,6 +93,10 @@ puts a random ID, fresh per connection, in front of it to name the publisher. A
 anonymous hop at any depth ranks below every fully identified route, whatever
 the costs say; among anonymous routes, cost keeps ordering.
 
+A relay operator can lower the [hop-path cap](/bin/relay/cluster#topology) to
+limit the detours explored after a withdrawal. It drops longer paths while
+preserving route selection among those within the cap.
+
 On moq-lite 07, an announcement may copy the head of its path and the tail of
 its relay chain from one still live on the same stream, so many broadcasts
 from a few origins behind the same relays stop repeating those bytes. Rust

rs/moq-net/benches/origin.rs

@@ -9,6 +9,10 @@
 //! down to its prefix and beneath it. Competing routes at that prefix are
 //! ranked once per change, then each cursor selects its first visible entry.
 //!
+//! A withdrawal wave is swept over the broadcasts it withdraws and the cursors
+//! polling through it, and a route churning beneath an unchanged winner over the
+//! table and its cursors.
+//!
 //! An equal-cost pool is swept the same way, over its members and the paths it
 //! already serves, and over those paths and the cursors watching the pool.
 //!
@@ -171,6 +175,97 @@ fn bench_withdraw(c: &mut Criterion) {
 	group.finish();
 }
 
+/// Two peers carry the same `broadcasts` of one publisher, and both withdraw them
+/// a broadcast at a time, every cursor polled after each withdrawal as a session's
+/// task would be. The first peer's withdrawals queue held updates; the second's, in
+/// the opposite order, make each ready behind the held rest. A poll that walks what
+/// it holds back shows up as a slope over broadcasts, multiplied by the cursors.
+fn bench_withdraw_wave(c: &mut Criterion) {
+	let mut group = c.benchmark_group("origin/withdraw_wave");
+	group.sample_size(10);
+	let route = |chain: &[u64]| {
+		let mut hops = Hops::new();
+		for &hop in chain {
+			hops.push(Hop::new(hop).unwrap()).unwrap();
+		}
+		origin::Route::default().with_hops(hops)
+	};
+	for broadcasts in [1_000, 10_000] {
+		for subscribers in [1, 20, 100] {
+			let id = BenchmarkId::from_parameter(format!("{broadcasts}b_{subscribers}s"));
+			group.throughput(Throughput::Elements((2 * broadcasts * subscribers) as u64));
+			group.bench_function(id, |b| {
+				b.iter_batched(
+					|| {
+						let (producer, mut driver) = origin::Producer::new(origin::Config::default());
+						// The holds read the driver's last clock reading, which stays put below.
+						driver
+							.poll(moq_net::time::Instant::now(), &kio::Waiter::noop())
+							.unwrap();
+						let near: Vec<_> = (0..broadcasts)
+							.map(|i| producer.dynamic(format!("room/{i}"), route(&[9, 2])).unwrap())
+							.collect();
+						let far: Vec<_> = (0..broadcasts)
+							.map(|i| producer.dynamic(format!("room/{i}"), route(&[9, 3, 4])).unwrap())
+							.collect();
+						let mut cursors: Vec<_> = (0..subscribers).map(|_| producer.consume().announced()).collect();
+						for cursor in &mut cursors {
+							while cursor.try_next().is_some() {}
+						}
+						(producer, driver, near, far, cursors)
+					},
+					|(producer, driver, near, far, mut cursors)| {
+						for route in near {
+							drop(route);
+							for cursor in &mut cursors {
+								assert!(cursor.try_next().is_none(), "the update is held");
+							}
+						}
+						for route in far.into_iter().rev() {
+							drop(route);
+							for cursor in &mut cursors {
+								cursor.try_next().expect("retract delivered");
+							}
+						}
+						(producer, driver, cursors)
+					},
+					BatchSize::LargeInput,
+				);
+			});
+		}
+	}
+	group.finish();
+}
+
+/// A route that never wins arriving at one broadcast's prefix and leaving again,
+/// with `broadcasts` in the table and `subscribers` cursors watching. Each change
+/// revisits every cursor presenting the prefix and finds its winner unchanged,
+/// so a cost per cursor shows up as a slope over subscribers, and one per route
+/// in the table as a slope over broadcasts.
+fn bench_route_churn(c: &mut Criterion) {
+	let mut group = c.benchmark_group("origin/route_churn");
+	for broadcasts in [1_000, 10_000] {
+		for subscribers in [1, 20, 100] {
+			let id = BenchmarkId::from_parameter(format!("{broadcasts}b_{subscribers}s"));
+			group.bench_function(id, |b| {
+				let (producer, _driver) = origin::Producer::new(origin::Config::default());
+				let _routes: Vec<_> = (0..broadcasts)
+					.map(|i| producer.dynamic(format!("room/{i}"), peer_route(1, 1)).unwrap())
+					.collect();
+				let mut cursors: Vec<_> = (0..subscribers).map(|_| producer.consume().announced()).collect();
+				for cursor in &mut cursors {
+					while cursor.try_next().is_some() {}
+				}
+				b.iter(|| drop(producer.dynamic("room/0", peer_route(2, 5)).unwrap()));
+				for cursor in &mut cursors {
+					assert!(cursor.try_next().is_none(), "the winner is unchanged");
+				}
+			});
+		}
+	}
+	group.finish();
+}
+
 /// `bench_announce` read through mounts: `publishers` routes under the fleet-wide
 /// `.svc/p0`, watched by `subscribers` project sessions that each mount it at
 /// their own `<project>/.svc`. Every mount aliases the one target, the worst
@@ -1298,6 +1393,8 @@ criterion_group!(
 	bench_announce,
 	bench_pending_churn,
 	bench_withdraw,
+	bench_withdraw_wave,
+	bench_route_churn,
 	bench_announce_mounted,
 	bench_announce_fleet,
 	bench_announce_duplicate,

rs/moq-net/src/client.rs

@@ -195,6 +195,7 @@ impl Client {
 
 		let start = lite::start(lite::Config {
 			runtime: runtime.clone(),
+			stats: self.stats.clone(),
 			session: session.clone(),
 			setup_stream: None,
 			publish,
@@ -272,6 +273,7 @@ impl Client {
 				// We advertise the request path in our SETUP for URL-less transports.
 				let (protocol, goaway) = ietf::start(ietf::Config {
 					runtime: runtime.clone(),
+					stats: self.stats.clone(),
 					session: session.clone(),
 					setup: None,
 					request_id_max: None,
@@ -348,7 +350,7 @@ impl Client {
 			Some(p) => return Err(Error::UnknownAlpn(p.to_string())),
 		};
 
-		let mut stream = Stream::open(&mut session, encoding).await?;
+		let mut stream = Stream::open(&mut session, encoding).await?.with_stats(&self.stats);
 
 		// The encoding is always an IETF version for SETUP negotiation.
 		let ietf_encoding = ietf::Version::try_from(encoding).map_err(|_| Error::Version)?;
@@ -387,6 +389,7 @@ impl Client {
 				let stream = stream.with_version(v);
 				let start = lite::start(lite::Config {
 					runtime: runtime.clone(),
+					stats: self.stats.clone(),
 					session: session.clone(),
 					setup_stream: Some(stream),
 					publish: publish.clone(),
@@ -423,6 +426,7 @@ impl Client {
 				// Draft 14-16: the path rode in the bidi SETUP above, not the uni one.
 				let (protocol, goaway) = ietf::start(ietf::Config {
 					runtime: runtime.clone(),
+					stats: self.stats.clone(),
 					session: session.clone(),
 					setup: Some(stream),
 					request_id_max,

rs/moq-net/src/coding/decode.rs

@@ -7,6 +7,17 @@ use thiserror::Error;
 pub trait Decode<V>: Sized {
 	/// Decode the value from the given buffer.
 	fn decode<B: bytes::Buf>(buf: &mut B, version: V) -> Result<Self, DecodeError>;
+
+	/// The control metric class of this decoded value, or `None` for selectors and media headers.
+	fn control(&self, _version: V) -> Option<crate::stats::Control> {
+		None
+	}
+
+	/// Leading bytes of this value left out of its [`Self::control`] count: a type selector
+	/// the sender writes apart from the message, and so never counts.
+	fn uncounted(&self) -> usize {
+		0
+	}
 }
 
 /// A decode error.

rs/moq-net/src/coding/encode.rs

@@ -50,6 +50,11 @@ pub trait Encode<V>: Sized {
 	/// Encode the value to the given writer.
 	fn encode<W: bytes::BufMut>(&self, w: &mut W, version: V) -> Result<(), EncodeError>;
 
+	/// The control metric class of this value, or `None` for selectors and media headers.
+	fn control(&self, _version: V) -> Option<crate::stats::Control> {
+		None
+	}
+
 	/// Encode the value into a [Bytes] buffer.
 	///
 	/// NOTE: This will allocate.