From bd2b703f92b650c079e9fdb556b3154eee0eca7f Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 4 Oct 2026 20:42:47 -0700
Subject: [PATCH 1/8] quest: claim quest/m1/cluster-routing/transit.md

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

From c2892d365d3db6245778c25d2903353e6bce7a94 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 4 Oct 2026 20:51:28 -0700
Subject: [PATCH 2/8] feat(relay): mark cluster links upstream

An upstream link is never offered a route learned on another upstream
link. The mark comes from a peer entry's upstream field when this relay
dials, and from the grant's upstream claim when it accepts.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-auth/src/error.rs       |   3 +
 rs/moq-auth/src/grant.rs       |  27 ++++-
 rs/moq-net/src/model/origin.rs | 144 ++++++++++++++++++++++----
 rs/moq-relay/src/auth.rs       |  24 ++++-
 rs/moq-relay/src/cluster.rs    | 178 ++++++++++++++++++++++++---------
 5 files changed, 302 insertions(+), 74 deletions(-)

diff --git a/rs/moq-auth/src/error.rs b/rs/moq-auth/src/error.rs
index 6825f06308..3a6e154f35 100644
--- a/rs/moq-auth/src/error.rs
+++ b/rs/moq-auth/src/error.rs
@@ -99,6 +99,9 @@ pub enum Error {
 	#[error("grant names nothing; the session is refused")]
 	UselessGrant,
 
+	#[error("grant marks an upstream that is not a peer")]
+	UpstreamWithoutPeer,
+
 	#[error("grant asks to be revalidated but never expires")]
 	UnboundedRevalidate,
 
diff --git a/rs/moq-auth/src/grant.rs b/rs/moq-auth/src/grant.rs
index bb414a56c0..9142d58664 100644
--- a/rs/moq-auth/src/grant.rs
+++ b/rs/moq-auth/src/grant.rs
@@ -49,6 +49,12 @@ pub struct Grant {
 	/// the cluster elsewhere, not here.
 	#[serde(skip_serializing_if = "std::ops::Not::not")]
 	pub peer: bool,
+
+	/// The cluster peer is upstream: the relay never offers it a route learned
+	/// from another upstream peer, so it never carries traffic between two.
+	/// Requires `peer`.
+	#[serde(skip_serializing_if = "std::ops::Not::not")]
+	pub upstream: bool,
 }
 
 impl Grant {
@@ -69,12 +75,15 @@ impl Grant {
 		tokio::time::Instant::now().checked_add(remaining)
 	}
 
-	/// Refuse a grant that admits nothing, asks to be revalidated without a bound or
-	/// at no interval, or has already expired.
+	/// Refuse a grant that admits nothing, marks a non-peer upstream, asks to be
+	/// revalidated without a bound or at no interval, or has already expired.
 	pub fn validate(&self) -> crate::Result<()> {
 		if self.publish.is_empty() && self.subscribe.is_empty() {
 			return Err(crate::Error::UselessGrant);
 		}
+		if self.upstream && !self.peer {
+			return Err(crate::Error::UpstreamWithoutPeer);
+		}
 		if self.revalidate.is_some() && self.expires.is_none() {
 			return Err(crate::Error::UnboundedRevalidate);
 		}
@@ -108,12 +117,14 @@ mod tests {
 			revalidate: Some(Duration::from_secs(60)),
 			tier: Some("websocket".into()),
 			peer: true,
+			upstream: true,
 		};
 		let json = serde_json::to_value(&grant).unwrap();
 		assert_eq!(json["expires"], 4_102_444_800_i64);
 		assert_eq!(json["revalidate"], 60);
 		assert_eq!(json["publish"], serde_json::json!(["alice/**"]));
 		assert_eq!(json["peer"], true);
+		assert_eq!(json["upstream"], true);
 		assert_eq!(serde_json::from_value::<Grant>(json).unwrap(), grant);
 	}
 
@@ -130,10 +141,11 @@ mod tests {
 			revalidate: Some(Duration::from_secs(60)),
 			tier: Some("websocket".into()),
 			peer: true,
+			upstream: true,
 		};
 		assert_eq!(
 			serde_json::to_string(&grant).unwrap(),
-			r#"{"publish":["alice/**"],"subscribe":["**"],"root":"pid/room","expires":4102444800,"revalidate":60,"tier":"websocket","peer":true}"#
+			r#"{"publish":["alice/**"],"subscribe":["**"],"root":"pid/room","expires":4102444800,"revalidate":60,"tier":"websocket","peer":true,"upstream":true}"#
 		);
 	}
 
@@ -177,6 +189,15 @@ mod tests {
 		assert!(matches!(grant.validate(), Err(crate::Error::ZeroRevalidate)));
 	}
 
+	#[test]
+	fn validate_refuses_an_upstream_that_is_not_a_peer() {
+		let mut grant = Grant::new(patterns(&["**"]), Patterns::new());
+		grant.upstream = true;
+		assert!(matches!(grant.validate(), Err(crate::Error::UpstreamWithoutPeer)));
+		grant.peer = true;
+		grant.validate().unwrap();
+	}
+
 	#[cfg(feature = "tokio")]
 	#[tokio::test(start_paused = true)]
 	async fn deadline_is_the_exact_expiry() {
diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs
index 00cd050446..040198ea1c 100644
--- a/rs/moq-net/src/model/origin.rs
+++ b/rs/moq-net/src/model/origin.rs
@@ -858,9 +858,8 @@ struct RouteEntry {
 	/// Whether this is a broadcast published on this origin, which wins a cost tie and
 	/// keeps its own cache.
 	local: bool,
-	/// Whether a handle marked [`Producer::peer`] inserted the entry, so it
-	/// entered from a cluster peer rather than here.
-	peer: bool,
+	/// The link the entry arrived on, from the handle that inserted it.
+	link: Link,
 	/// The queue requests under this route are served from, when the announcer
 	/// serves content on demand (a [`Dynamic`]). `None` for an advertise-only
 	/// announcement ([`Producer::announce`]) and for a local broadcast.
@@ -897,9 +896,9 @@ impl RouteEntry {
 
 	/// Where the entry entered this origin.
 	fn entered(&self) -> Source {
-		match self.peer {
-			true => Source::Peer(self.via),
-			false => Source::Local,
+		match self.link {
+			Link::Local => Source::Local,
+			Link::Peer | Link::Upstream => Source::Peer(self.via),
 		}
 	}
 
@@ -977,15 +976,33 @@ struct Horizon {
 	exclude: Option<Hop>,
 	/// Hide the routes that entered from a cluster peer ([`Consumer::local`]).
 	local: bool,
+	/// Hide the routes learned on an upstream link, because this reader is one
+	/// ([`Producer::upstream`]): a relay never transits between two upstreams.
+	upstream: bool,
 }
 
 impl Horizon {
 	/// Whether `entry` may be observed or served through this horizon.
 	fn admits(&self, entry: &RouteEntry) -> bool {
-		!(self.local && entry.peer) && entry.visible_to(self.exclude)
+		!(self.local && entry.link != Link::Local)
+			&& !(self.upstream && entry.link == Link::Upstream)
+			&& entry.visible_to(self.exclude)
 	}
 }
 
+/// The link a handle's routes arrive on; see [`Producer::peer`] and
+/// [`Producer::upstream`].
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
+enum Link {
+	/// Here: an in-process producer or a client session.
+	#[default]
+	Local,
+	/// A cluster peer.
+	Peer,
+	/// A cluster peer marked upstream.
+	Upstream,
+}
+
 /// One remotely-served front in [`OriginState::fronts`]: the shared broadcast at a
 /// path plus the channel requesters resolve through.
 #[derive(Clone)]
@@ -1407,9 +1424,9 @@ pub struct Producer {
 	// session tagged this handle via [`Self::with_stats`].
 	stats: stats::Session,
 
-	// Whether routes announced through this handle entered from a cluster peer
-	// (see [`Self::peer`]).
-	peer: bool,
+	// The link routes announced through this handle arrive on (see
+	// [`Self::peer`] and [`Self::upstream`]).
+	link: Link,
 
 	// Submission handle to the origin's [`Driver`]: source watchers, fronts, and
 	// serve tasks queued here run when the driver is polled. Closed once the
@@ -1441,7 +1458,7 @@ impl Producer {
 			pool: config.pool,
 			cache_duration: config.cache_duration,
 			stats: stats::Session::default(),
-			peer: false,
+			link: Link::Local,
 			tasks,
 			timers: timers.clone(),
 		};
@@ -1473,7 +1490,22 @@ impl Producer {
 	/// entered here from what a peer forwarded. The hop chain cannot: a client
 	/// and a peer each append one hop.
 	pub fn peer(mut self) -> Self {
-		self.peer = true;
+		if self.link == Link::Local {
+			self.link = Link::Peer;
+		}
+		self
+	}
+
+	/// Mark this handle (and any handle derived from it) as an upstream peer's:
+	/// a [`Self::peer`] whose routes are never offered to another upstream.
+	/// A [`Consumer`] taken from it hides every route an upstream handle
+	/// announced, so this origin never carries traffic between two upstreams.
+	///
+	/// Hand it to a session with a peer that should reach everything here
+	/// without this relay carrying its traffic to other upstreams, such as an
+	/// edge's link to a core or a drone's link to its CDN.
+	pub fn upstream(mut self) -> Self {
+		self.link = Link::Upstream;
 		self
 	}
 
@@ -1508,7 +1540,7 @@ impl Producer {
 			pool: cache::Pool::default(),
 			cache_duration: Duration::MAX,
 			stats: stats::Session::default(),
-			peer: false,
+			link: Link::Local,
 			tasks,
 			timers: Clock::default(),
 		}
@@ -1575,7 +1607,7 @@ impl Producer {
 			prefixes: vec![(full.clone(), claim)],
 			scope: self.scope.allowed.clone(),
 			local: true,
-			peer: self.peer,
+			link: self.link,
 			stats: self.stats.clone(),
 		};
 		let info = broadcast::Info {
@@ -1731,7 +1763,7 @@ impl Producer {
 			pool: self.pool.clone(),
 			cache_duration: self.cache_duration,
 			stats: self.stats.clone(),
-			peer: self.peer,
+			link: self.link,
 			tasks: self.tasks.clone(),
 			timers: self.timers.clone(),
 		})
@@ -1843,8 +1875,8 @@ struct Announcing {
 	/// The absolute paths the producer is authorized to serve.
 	scope: Patterns,
 	local: bool,
-	/// Whether the producer was marked [`Producer::peer`].
-	peer: bool,
+	/// The link the producer's routes arrive on.
+	link: Link,
 	stats: stats::Session,
 }
 
@@ -1866,7 +1898,7 @@ impl Announcing {
 			prefixes: vec![(requested, claim)],
 			scope: producer.scope.allowed.clone(),
 			local: false,
-			peer: producer.peer,
+			link: producer.link,
 			stats: producer.stats.clone(),
 		})
 	}
@@ -1898,7 +1930,7 @@ impl Announcing {
 				cost: route.cost,
 				via,
 				local: self.local,
-				peer: self.peer,
+				link: self.link,
 				server: serving.server.clone(),
 				source: serving.source.clone(),
 				advertised: serving.advertised,
@@ -3827,7 +3859,8 @@ pub struct Consumer {
 	// Split horizon: routes whose hop chain or announcing session (`via`) is the
 	// excluded peer are invisible to `announced` and skipped by
 	// `request_broadcast`, so a peer is never served (or advertised) its own
-	// content back. A local view (`Self::local`) hides peer routes the same way.
+	// content back. A local view (`Self::local`) hides peer routes the same way,
+	// and a view taken from an upstream handle hides upstream routes.
 	horizon: Horizon,
 
 	// Which routes beneath a hidden (`.`-prefixed) segment `announced` reports.
@@ -3856,7 +3889,10 @@ impl Consumer {
 			root: producer.root.clone(),
 			shared: producer.shared.clone(),
 			stats,
-			horizon: Horizon::default(),
+			horizon: Horizon {
+				upstream: producer.link == Link::Upstream,
+				..Horizon::default()
+			},
 			hidden: Hidden::default(),
 			pool: producer.pool.clone(),
 			cache_duration: producer.cache_duration,
@@ -3883,7 +3919,7 @@ impl Consumer {
 	}
 
 	/// A view of the routes that entered here: every route a handle marked
-	/// [`Producer::peer`] announced is hidden from [`Self::announced`] and never
+	/// [`Producer::peer`] or [`Producer::upstream`] announced is hidden from [`Self::announced`] and never
 	/// resolved by [`Self::request_broadcast`].
 	///
 	/// On a relay, this is what the relay ingests itself, from clients and
@@ -6815,6 +6851,70 @@ mod tests {
 		);
 	}
 
+	/// An upstream link is offered what entered here and what other peers
+	/// forwarded, but never a route learned on another upstream link, and the
+	/// best route it sees skips a better upstream one.
+	#[tokio::test]
+	async fn upstream_view_hides_upstream_routes() {
+		let producer = origin(1).produce();
+		let peer = producer.clone().peer();
+		let core = producer.clone().upstream();
+		let mut toward_core = core.consume().announced();
+		let mut everyone = producer.consume().announced();
+
+		let _core = core
+			.dynamic("core", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
+			.unwrap();
+		assert_eq!(everyone.assert_next_active("core").source(), Source::Peer(origin(7)));
+		toward_core.assert_next_wait();
+
+		let _mesh = peer
+			.dynamic("mesh", Route::default().with_hops(hops(&[8])).with_via(origin(8)))
+			.unwrap();
+		assert_eq!(toward_core.assert_next_active("mesh").source(), Source::Peer(origin(8)));
+
+		// A path both an upstream and a mesh peer reach: the upstream route is
+		// cheaper, yet the upstream view selects the mesh peer's.
+		let _cheap = core
+			.dynamic("both", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
+			.unwrap();
+		let mesh = peer
+			.dynamic(
+				"both",
+				Route::default().with_hops(hops(&[8])).with_via(origin(8)).with_cost(9),
+			)
+			.unwrap();
+		assert_eq!(toward_core.assert_next_active("both").source(), Source::Peer(origin(8)));
+		drop(mesh);
+		toward_core.assert_next_ended("both");
+
+		// `peer` never demotes an upstream handle.
+		let _still = core
+			.clone()
+			.peer()
+			.dynamic("again", Route::default().with_hops(hops(&[7])).with_via(origin(7)))
+			.unwrap();
+		toward_core.assert_next_wait();
+
+		// Resolution agrees with the cursor: an upstream-only path is unroutable
+		// toward an upstream, while the full view queues it on the core's route.
+		let err = core
+			.consume()
+			.request_broadcast("core/alice")
+			.now_or_never()
+			.expect("unroutable")
+			.err()
+			.unwrap();
+		assert!(matches!(err, Error::Unroutable));
+		assert!(
+			producer
+				.consume()
+				.request_broadcast("core/alice")
+				.now_or_never()
+				.is_none()
+		);
+	}
+
 	/// A handler that rejects a path with `Unroutable` while its route stands
 	/// gives the requester that answer; the front must not re-ask the same route
 	/// forever, which would spin the origin driver.
diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs
index f3cee4285c..8e85251ad7 100644
--- a/rs/moq-relay/src/auth.rs
+++ b/rs/moq-relay/src/auth.rs
@@ -385,6 +385,8 @@ pub struct Token {
 	pub tier: Tier,
 	/// Whether the session is a cluster peer, so its routes entered elsewhere.
 	pub peer: bool,
+	/// Whether the peer is upstream, so it is never offered another upstream's routes.
+	pub upstream: bool,
 }
 
 impl Token {
@@ -407,6 +409,7 @@ impl Token {
 			publish: grant.publish.clone(),
 			tier: crate::configured_tier(grant.tier.clone()),
 			peer: grant.peer,
+			upstream: grant.upstream,
 		}
 	}
 
@@ -480,8 +483,9 @@ impl Lease {
 	///
 	/// A changed root or mounts, or a narrower grant, ends it: origin handles cannot yet narrow
 	/// a live scope in place (tracked by `quest/m1/auth/narrowing.md`). A flipped
-	/// `peer` ends it too, since the routes it already announced would be
-	/// misreported as entering here or from a peer. A changed tier keeps the
+	/// `peer` or `upstream` ends it too, since the routes it already announced
+	/// would be misreported as entering here or from a peer, or offered to the
+	/// wrong links. A changed tier keeps the
 	/// session and moves its [stats](Self::with_stats) to the new tier.
 	pub async fn ended(&mut self) -> lease::Reason {
 		loop {
@@ -506,6 +510,9 @@ impl Lease {
 						if fresh.peer != self.token.peer {
 							return "peer changed".into();
 						}
+						if fresh.upstream != self.token.upstream {
+							return "upstream changed".into();
+						}
 						if !self.token.covered_by(&fresh) {
 							return lease::Reason::Narrowed;
 						}
@@ -1054,7 +1061,7 @@ mod tests {
 	/// Routes a session announced were recorded as a peer's or not; a re-check that
 	/// flips it closes the session rather than misreport them.
 	#[tokio::test]
-	async fn a_recheck_that_flips_peer_closes() {
+	async fn a_recheck_that_flips_the_link_closes() {
 		let grant = Grant::new(patterns(&["**"]), patterns(&["**"]));
 		let (producer, consumer) = lease::Producer::new(grant.clone());
 		let mut lease = Lease::new("/", consumer);
@@ -1062,8 +1069,17 @@ mod tests {
 
 		let mut peer = grant;
 		peer.peer = true;
-		producer.update(peer);
+		producer.update(peer.clone());
 		assert_eq!(lease.ended().await.to_string(), "peer changed");
+
+		// Routes already offered to (or withheld from) the session were filtered
+		// by whether it is upstream, so flipping that closes it too.
+		let (producer, consumer) = lease::Producer::new(peer.clone());
+		let mut lease = Lease::new("/", consumer);
+		let mut upstream = peer;
+		upstream.upstream = true;
+		producer.update(upstream);
+		assert_eq!(lease.ended().await.to_string(), "upstream changed");
 	}
 
 	#[test]
diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs
index 1c4c78f70b..184365a434 100644
--- a/rs/moq-relay/src/cluster.rs
+++ b/rs/moq-relay/src/cluster.rs
@@ -34,9 +34,10 @@ const CONNECT_API_POLL_INTERVAL: Duration = Duration::from_secs(30);
 ///
 /// Accepts a URL string or an object with `url` plus
 /// policy: `cost` prices the link, `egress` declares this relay's own price to
-/// the peer, and `token` carries the peer's credential. `egress` defaults to
-/// `cost`; a different effective value is rejected until asymmetric routing
-/// lands. An object `token` behaves exactly like an inline `?jwt=` and is
+/// the peer, `token` carries the peer's credential, and `upstream` marks the
+/// link upstream, so it is never offered a route learned on another upstream
+/// link. `egress` defaults to `cost`; a different effective value is rejected
+/// until asymmetric routing lands. An object `token` behaves exactly like an inline `?jwt=` and is
 /// redacted the same way. Unknown object fields are rejected, as is an object
 /// that sets policy while its `url` still carries `?cost=` or `?jwt=`.
 #[derive(Clone, PartialEq, Eq)]
@@ -46,6 +47,7 @@ pub struct Peer {
 	cost: Option<u64>,
 	egress: Option<u64>,
 	token: Option<String>,
+	upstream: bool,
 }
 
 impl std::fmt::Debug for Peer {
@@ -57,6 +59,7 @@ impl std::fmt::Debug for Peer {
 			.field("cost", &self.cost)
 			.field("egress", &self.egress)
 			.field("token", &self.token.as_ref().map(|_| "..."))
+			.field("upstream", &self.upstream)
 			.finish()
 	}
 }
@@ -69,6 +72,7 @@ impl Peer {
 			cost: None,
 			egress: None,
 			token: None,
+			upstream: false,
 		}
 	}
 
@@ -94,6 +98,12 @@ impl Peer {
 		self.token.as_deref()
 	}
 
+	/// Whether the link is upstream: it is never offered a route learned on
+	/// another upstream link.
+	pub fn upstream(&self) -> bool {
+		self.upstream
+	}
+
 	/// Price the link this peer is dialed on.
 	pub fn with_cost(mut self, cost: u64) -> Self {
 		self.cost = Some(cost);
@@ -111,6 +121,12 @@ impl Peer {
 		self.token = Some(token.into());
 		self
 	}
+
+	/// Mark the link upstream.
+	pub fn with_upstream(mut self, upstream: bool) -> Self {
+		self.upstream = upstream;
+		self
+	}
 }
 
 impl std::str::FromStr for Peer {
@@ -134,6 +150,8 @@ struct PeerObject {
 	egress: Option<u64>,
 	#[serde(default)]
 	token: Option<String>,
+	#[serde(default)]
+	upstream: bool,
 }
 
 /// A bare string is the URL form; a map is the object form. Dispatching on the
@@ -150,7 +168,7 @@ impl<'de> serde::Deserialize<'de> for Peer {
 			type Value = Peer;
 
 			fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-				f.write_str("a peer URL string or an object with url, cost, egress, token")
+				f.write_str("a peer URL string or an object with url, cost, egress, token, upstream")
 			}
 
 			fn visit_str<E: serde::de::Error>(self, url: &str) -> Result<Peer, E> {
@@ -165,6 +183,7 @@ impl<'de> serde::Deserialize<'de> for Peer {
 					cost: object.cost,
 					egress: object.egress,
 					token: object.token,
+					upstream: object.upstream,
 				})
 			}
 		}
@@ -178,7 +197,7 @@ impl serde::Serialize for Peer {
 	where
 		S: serde::Serializer,
 	{
-		if self.cost.is_none() && self.egress.is_none() && self.token.is_none() {
+		if self.cost.is_none() && self.egress.is_none() && self.token.is_none() && !self.upstream {
 			return serializer.serialize_str(&self.url);
 		}
 		use serde::ser::SerializeStruct as _;
@@ -192,6 +211,9 @@ impl serde::Serialize for Peer {
 		if self.token.is_some() {
 			len += 1;
 		}
+		if self.upstream {
+			len += 1;
+		}
 		let mut state = serializer.serialize_struct("Peer", len)?;
 		state.serialize_field("url", &self.url)?;
 		if let Some(cost) = &self.cost {
@@ -203,6 +225,9 @@ impl serde::Serialize for Peer {
 		if let Some(token) = &self.token {
 			state.serialize_field("token", token)?;
 		}
+		if self.upstream {
+			state.serialize_field("upstream", &true)?;
+		}
 		state.end()
 	}
 }
@@ -222,6 +247,8 @@ struct DialTarget {
 	/// Present the mDNS credential on `/.cluster/<credential>` and skip
 	/// [`Config::token`]. LAN only.
 	lan: bool,
+	/// Never offer this peer a route learned on another upstream link.
+	upstream: bool,
 }
 
 impl DialTarget {
@@ -276,6 +303,7 @@ impl DialTarget {
 			cost,
 			fingerprint: None,
 			lan: false,
+			upstream: peer.upstream,
 		})
 	}
 
@@ -310,6 +338,7 @@ impl DialTarget {
 			cost,
 			fingerprint: peer.fingerprint.clone(),
 			lan: true,
+			upstream: false,
 		})
 	}
 }
@@ -555,7 +584,7 @@ pub struct Config {
 
 	/// Connect to one or more other cluster nodes. Each entry is a full URL, e.g.
 	/// `https://host/?jwt=TOKEN`, or an object with `url`, `cost`, `egress`,
-	/// and `token`; see [`Peer`]. A bare host or `host:port` is refused; pass the
+	/// `token`, and `upstream`; see [`Peer`]. A bare host or `host:port` is refused; pass the
 	/// full URL instead. Accepts a comma-separated list on the CLI or repeat the
 	/// flag; in config files use a TOML array of URLs and/or objects.
 	///
@@ -579,7 +608,7 @@ pub struct Config {
 	/// Fetch the list of peers to dial from an HTTP(S) URL or a local file,
 	/// reloading at runtime without a restart. The source returns a JSON array
 	/// of peers: bare URL strings and/or objects with `url`, `cost`, `egress`,
-	/// and `token`, exactly as [`Self::connect`] accepts, e.g.
+	/// `token`, and `upstream`, exactly as [`Self::connect`] accepts, e.g.
 	/// `["https://a.pop.example/?cost=1", {"url": "https://b.pop.example/", "cost": 2}]`.
 	/// An http(s) URL is re-checked on a fixed cadence, with caching, conditional revalidation
 	/// (`ETag` / `Last-Modified`), and stale-if-error handled by the shared HTTP
@@ -1158,26 +1187,29 @@ impl Cluster {
 	/// Returns an [`origin::Producer`] scoped to this session's subscribe permissions.
 	///
 	/// Passed by reference to [`moq_net::Server::with_publisher`] (or the
-	/// equivalent per-request setter), which derives the read handle.
+	/// equivalent per-request setter), which derives the read handle. On an
+	/// upstream grant that handle never sees a route learned from another upstream.
 	pub fn subscriber(&self, token: &auth::Token) -> Option<origin::Producer> {
 		self.mounted(token)?.scope(&token.root, &token.subscribe).ok()
 	}
 
 	/// Returns an [`origin::Producer`] scoped to this session's publish permissions,
-	/// marked [`origin::Producer::peer`] when the grant names a cluster peer.
+	/// marked [`origin::Producer::peer`] when the grant names a cluster peer and
+	/// [`origin::Producer::upstream`] when that peer is upstream.
 	/// Nothing is published beneath the grant's mounts.
 	pub fn publisher(&self, token: &auth::Token) -> Option<origin::Producer> {
-		let publisher = self.mounted(token)?.scope(&token.root, &token.publish).ok()?;
-		Some(match token.peer {
-			true => publisher.peer(),
-			false => publisher,
-		})
+		self.mounted(token)?.scope(&token.root, &token.publish).ok()
 	}
 
-	/// The origin with the grant's mounts applied, before it is scoped to the
-	/// session. A mount the origin refuses (overlapping another) admits nothing.
+	/// The origin marked with the grant's link and with its mounts applied,
+	/// before it is scoped to the session. A mount the origin refuses
+	/// (overlapping another) admits nothing.
 	fn mounted(&self, token: &auth::Token) -> Option<origin::Producer> {
-		let mut origin = self.origin.clone();
+		let mut origin = match (token.upstream, token.peer) {
+			(true, _) => self.origin.clone().upstream(),
+			(false, true) => self.origin.clone().peer(),
+			(false, false) => self.origin.clone(),
+		};
 		for (at, target) in &token.mounts {
 			origin = match origin.mount(token.root.join(at), target) {
 				Ok(origin) => origin,
@@ -1634,7 +1666,6 @@ impl Cluster {
 	#[tracing::instrument("remote", skip_all, err, fields(remote = %target.key))]
 	async fn run_remote(self, target: &DialTarget, token: String) -> anyhow::Result<()> {
 		let mut urls = target.addrs();
-		let cost = target.cost;
 		// Apply the shared cluster token unless the URL already carries its own
 		// non-empty `?jwt=` (a per-peer inline token or object `token` wins; the
 		// shared token still covers peers that have none). An empty
@@ -1661,9 +1692,7 @@ impl Cluster {
 
 		loop {
 			let started = tokio::time::Instant::now();
-			let result = self
-				.run_remote_once(&addrs, cost, target.lan, target.fingerprint.as_deref())
-				.await;
+			let result = self.run_remote_once(&addrs, target).await;
 			let elapsed = started.elapsed();
 
 			match result {
@@ -1682,37 +1711,25 @@ impl Cluster {
 		}
 	}
 
-	async fn run_remote_once(
-		&self,
-		addrs: &moq_tokio::Addrs,
-		cost: Option<u64>,
-		lan: bool,
-		fingerprint: Option<&str>,
-	) -> anyhow::Result<()> {
+	/// One session to `target` over `addrs`, which carry any credential.
+	async fn run_remote_once(&self, addrs: &moq_tokio::Addrs, target: &DialTarget) -> anyhow::Result<()> {
 		// Each attempt is its own session, so it gets its own id. Matches the span an
 		// accepted connection runs under, so both directions log the same way.
 		let id = self.next_connection_id();
-		self.run_remote_session(id, addrs, cost, lan, fingerprint)
+		self.run_remote_session(id, addrs, target)
 			.instrument(tracing::info_span!("conn", id))
 			.await
 	}
 
-	async fn run_remote_session(
-		&self,
-		id: u64,
-		addrs: &moq_tokio::Addrs,
-		cost: Option<u64>,
-		lan: bool,
-		fingerprint: Option<&str>,
-	) -> anyhow::Result<()> {
+	async fn run_remote_session(&self, id: u64, addrs: &moq_tokio::Addrs, target: &DialTarget) -> anyhow::Result<()> {
 		// The peer URL may carry the cluster JWT in its query, so neither the log
 		// line nor the node label below may show the raw URL.
 		let first = addrs.as_slice().first().expect("Addrs is non-empty").url();
 		let redacted = moq_tokio::RedactedUrl::new(first);
 		tracing::info!(url = %redacted, "dialing cluster peer");
 
-		let client = if lan {
-			self.lan_client(fingerprint)?
+		let client = if target.lan {
+			self.lan_client(target.fingerprint.as_deref())?
 		} else {
 			self.client
 				.clone()
@@ -1721,14 +1738,18 @@ impl Cluster {
 
 		// Cluster dials use their configured stats tier. Cluster peers carry no auth
 		// root, so presence is keyed under the empty root within the cluster tier.
-		// The peer's routes entered the cluster elsewhere. A peer that predates the
-		// hidden opt-in still discovers our hidden routes; see `Cluster::scope`.
-		let origin = self.origin.clone().peer();
+		// The peer's routes entered the cluster elsewhere, and an upstream peer is
+		// never offered another upstream's. A peer that predates the hidden opt-in
+		// still discovers our hidden routes; see `Cluster::scope`.
+		let origin = match target.upstream {
+			true => self.origin.clone().upstream(),
+			false => self.origin.clone().peer(),
+		};
 		let mut client = client
 			.with_publisher(origin.consume().with_hidden(true))
 			.with_subscriber(origin)
 			.with_stats(self.stats.tier(self.cluster_tier()).session(""));
-		if let Some(cost) = cost {
+		if let Some(cost) = target.cost {
 			client = client.with_cost(cost);
 		}
 		// The GOAWAY lifecycle lives in the reconnect loop: on an upstream GOAWAY it
@@ -2160,6 +2181,7 @@ mod tests {
 			cost: None,
 			fingerprint: None,
 			lan: false,
+			upstream: false,
 		}
 	}
 
@@ -2435,6 +2457,28 @@ mod tests {
 		);
 	}
 
+	/// `upstream` is object-only policy: it round-trips, reaches the dial target,
+	/// and flipping it is a different configuration, so the API redials.
+	#[test]
+	fn peer_object_upstream() {
+		let object: Peer = serde_json::from_str(r#"{"url": "https://core.example/", "upstream": true}"#)
+			.expect("parse upstream peer");
+		assert_eq!(object, Peer::new("https://core.example/").with_upstream(true));
+		assert_eq!(
+			serde_json::to_string(&object).unwrap(),
+			r#"{"url":"https://core.example/","upstream":true}"#
+		);
+		let upstream = DialTarget::from_peer(&object).unwrap();
+		assert!(upstream.upstream);
+		let plain = DialTarget::from_peer(&Peer::new("https://core.example/")).unwrap();
+		assert_eq!(upstream.key, plain.key);
+		assert_ne!(upstream, plain);
+		assert_eq!(
+			serde_json::to_string(&Peer::new("https://core.example/")).unwrap(),
+			r#""https://core.example/""#
+		);
+	}
+
 	/// `egress` defaults to `cost`, so a matching value is accepted and prices
 	/// the link exactly like the bare URL form. An unpriced link costs 1, so
 	/// `egress = 1` alone is symmetric too.
@@ -2733,7 +2777,7 @@ mod tests {
 		// that mutate it.
 		let _env = crate::test_env::EnvGuard::lock();
 
-		let toml = "[cluster]\nconnect = [\"https://a.example/?cost=1\", { url = \"https://b.example/\", cost = 2, egress = 2, token = \"secret\" }]\n";
+		let toml = "[cluster]\nconnect = [\"https://a.example/?cost=1\", { url = \"https://b.example/\", cost = 2, egress = 2, token = \"secret\", upstream = true }]\n";
 		let dir = std::env::temp_dir().join("moq-relay-cluster-test");
 		std::fs::create_dir_all(&dir).unwrap();
 		let path = dir.join("cluster-connect-object-toml.toml");
@@ -2748,7 +2792,8 @@ mod tests {
 				Peer::new("https://b.example/")
 					.with_cost(2)
 					.with_egress(2)
-					.with_token("secret"),
+					.with_token("secret")
+					.with_upstream(true),
 			]
 		);
 		// The object normalizes exactly like its URL spellings.
@@ -3122,6 +3167,7 @@ mod tests {
 			cost: None,
 			fingerprint: Some(fp),
 			lan: true,
+			upstream: false,
 		};
 		let _dial = fingerprint.dial_lan_target(&target).expect("dial");
 
@@ -3196,6 +3242,48 @@ mod tests {
 		assert!(try_next_announced(&mut local).is_none());
 	}
 
+	/// An upstream grant's session is a peer that never reads another upstream's
+	/// routes: two accepted cores never transit through this relay.
+	#[tokio::test]
+	async fn upstream_grant_hides_other_upstreams() {
+		let cluster = new_cluster(Config::default()).expect("cluster");
+		let all: moq_net::Patterns = [moq_net::Pattern::all()].into_iter().collect();
+
+		let client = auth::Token::new("/", &moq_auth::Grant::new(all.clone(), all.clone()));
+		let mut grant = moq_auth::Grant::new(all.clone(), all);
+		grant.peer = true;
+		let peer = auth::Token::new("/", &grant);
+		grant.upstream = true;
+		let core = auth::Token::new("/", &grant);
+
+		let _ingest = cluster
+			.publisher(&client)
+			.expect("client")
+			.publish("ingest", Default::default());
+		let _mesh = cluster
+			.publisher(&peer)
+			.expect("peer")
+			.publish("mesh", Default::default());
+		let _core = cluster
+			.publisher(&core)
+			.expect("core")
+			.publish("core", Default::default());
+
+		let mut everyone = cluster.subscriber(&client).expect("client").consume().announced();
+		let mut prefixes = std::iter::from_fn(|| try_next_announced(&mut everyone))
+			.map(|announce| announce.prefix.to_string())
+			.collect::<Vec<_>>();
+		prefixes.sort();
+		assert_eq!(prefixes, ["core", "ingest", "mesh"]);
+
+		let mut toward_core = cluster.subscriber(&core).expect("core").consume().announced();
+		let mut prefixes = std::iter::from_fn(|| try_next_announced(&mut toward_core))
+			.map(|announce| announce.prefix.to_string())
+			.collect::<Vec<_>>();
+		prefixes.sort();
+		assert_eq!(prefixes, ["ingest", "mesh"]);
+	}
+
 	/// A `/.cluster` request on a cluster without LAN discovery is refused.
 	#[cfg(feature = "cluster-lan")]
 	#[tokio::test]

From 4f22ef5a7c37d45270b4510e7188fcec406c4f9a Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 4 Oct 2026 21:00:36 -0700
Subject: [PATCH 3/8] test(relay): upstream links across tiered regions and a
 drone uplink

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-relay/tests/upstream_links.rs | 349 +++++++++++++++++++++++++++
 1 file changed, 349 insertions(+)
 create mode 100644 rs/moq-relay/tests/upstream_links.rs

diff --git a/rs/moq-relay/tests/upstream_links.rs b/rs/moq-relay/tests/upstream_links.rs
new file mode 100644
index 0000000000..b1333f4d55
--- /dev/null
+++ b/rs/moq-relay/tests/upstream_links.rs
@@ -0,0 +1,349 @@
+//! Upstream links over real TCP sessions: a relay never offers a route learned on
+//! one upstream link to another, whichever side dialed, and everything else
+//! still transits.
+//!
+//! Each relay is a [`cluster::Cluster`] serving accepted sessions through an
+//! embedded auth decider that reads the dialed path: `/upstream` admits the
+//! dialer as an upstream peer, `/peer` as a plain one.
+
+use std::collections::BTreeMap;
+use std::time::Duration;
+
+use moq_auth::{Grant, Pattern, lease};
+use moq_net::{announce, origin};
+use moq_relay::{
+	Connection, auth,
+	cluster::{self, Peer},
+};
+
+const TIMEOUT: Duration = Duration::from_secs(15);
+
+/// Bound `fut` by [`TIMEOUT`], naming the step that hung.
+async fn within<T>(step: &str, fut: impl std::future::Future<Output = T>) -> T {
+	tokio::time::timeout(TIMEOUT, fut)
+		.await
+		.unwrap_or_else(|_| panic!("timed out: {step}"))
+}
+
+/// Run on a big-stack thread: several relays' transport types make the test
+/// future large enough to overflow libtest's default stack, as in
+/// `goaway_cluster.rs`.
+fn run_cluster_test<F>(fut: F)
+where
+	F: std::future::Future<Output = ()> + Send + 'static,
+{
+	std::thread::Builder::new()
+		.stack_size(32 * 1024 * 1024)
+		.spawn(move || {
+			tokio::runtime::Builder::new_current_thread()
+				.enable_all()
+				.build()
+				.expect("build test runtime")
+				.block_on(fut);
+		})
+		.expect("spawn test thread")
+		.join()
+		.expect("test thread panicked");
+}
+
+/// One relay: its origin, listen port, and every task it runs. Dropping it
+/// aborts them all, closing its sessions like a crash.
+struct Node {
+	origin: origin::Producer,
+	port: u16,
+	_tasks: tokio::task::JoinSet<()>,
+}
+
+impl Node {
+	/// Dial this node from another. `upstream` is the dialer's own mark; `as_upstream`
+	/// asks this node to admit the dialer as its upstream.
+	fn dial(&self, upstream: bool, as_upstream: bool) -> Peer {
+		let path = if as_upstream { "upstream" } else { "peer" };
+		Peer::new(format!("tcp://127.0.0.1:{}/{path}", self.port)).with_upstream(upstream)
+	}
+}
+
+/// What the embedded decider grants a dial at `path`: everything, as a peer,
+/// upstream when asked.
+fn grant(path: &str) -> Grant {
+	let all: moq_net::Patterns = [Pattern::all()].into_iter().collect();
+	let mut grant = Grant::new(all.clone(), all);
+	grant.root = Some(String::new());
+	grant.peer = true;
+	grant.upstream = path == "/upstream";
+	grant
+}
+
+async fn node(id: u64, connect: Vec<Peer>) -> Node {
+	let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+	let mut tasks = tokio::task::JoinSet::new();
+
+	let mut listen = moq_tokio::listen::Config::default();
+	listen.tcp.bind = Some("127.0.0.1:0".parse().expect("parse addr"));
+	let mut listener = listen
+		.init(Default::default())
+		.expect("server init")
+		.listen()
+		.await
+		.expect("listen");
+	let port = listener.tcp_local_addr().expect("tcp listener bound").port();
+
+	let mut client = moq_tokio::connect::Config::default();
+	client.tls.insecure = Some(true);
+	let client = client.init(Default::default()).expect("client init");
+
+	let mut config = cluster::Config::default();
+	config.id = Some(id);
+	config.connect = connect;
+	let cluster = cluster::Cluster::new(cluster::Options::new(config))
+		.expect("cluster init")
+		.with_client(client);
+
+	let (auth, mut admissions) = auth::Auth::embedded(format!("node-{id}"));
+	tasks.spawn(async move {
+		while let Some(admission) = admissions.next().await {
+			let grant = grant(&admission.request.path);
+			admission.grant(lease::Consumer::fixed(grant));
+		}
+	});
+
+	let serving = cluster.clone();
+	tasks.spawn(async move {
+		let mut sessions = tokio::task::JoinSet::new();
+		while let Some(request) = listener.accept().await {
+			let connection = Connection::new(request, serving.clone(), auth.clone());
+			sessions.spawn(async move {
+				let _ = connection.run().await;
+			});
+		}
+	});
+
+	let started = cluster.clone().start().await.expect("cluster start");
+	tasks.spawn(async move {
+		let _ = started.run().await;
+	});
+
+	Node {
+		origin: cluster.origin.clone(),
+		port,
+		_tasks: tasks,
+	}
+}
+
+/// A broadcast writing a group every 50ms on `origin`, so a subscriber attaching
+/// at any point receives one. Dropping it ends the broadcast.
+struct Publisher {
+	_broadcast: moq_net::broadcast::Producer,
+	writer: tokio::task::AbortHandle,
+}
+
+impl Drop for Publisher {
+	fn drop(&mut self) {
+		self.writer.abort();
+	}
+}
+
+fn publish(origin: &origin::Producer, path: &str) -> Publisher {
+	let broadcast = origin.publish(path, Default::default()).expect("publish");
+	let track = broadcast.create_track("video", None).expect("create track");
+	let writer = tokio::spawn(async move {
+		loop {
+			let Ok(mut group) = track.append_group() else { break };
+			if group.write_frame(moq_net::Timestamp::ZERO, b"hello".as_ref()).is_err() || group.finish().is_err() {
+				break;
+			}
+			tokio::time::sleep(Duration::from_millis(50)).await;
+		}
+	})
+	.abort_handle();
+	Publisher {
+		_broadcast: broadcast,
+		writer,
+	}
+}
+
+/// Subscribe to `path` through `origin`'s routes and read one frame.
+async fn read_frame(origin: &origin::Producer, path: &str) {
+	let consumer = origin.consume();
+	let broadcast = within(&format!("route to {path}"), consumer.routed_broadcast(path))
+		.await
+		.unwrap_or_else(|err| panic!("{path} unroutable: {err}"));
+	let subscription = moq_net::track::Subscription::default().with_max_age(Duration::from_secs(1));
+	let mut track = within(
+		&format!("subscribe to {path}"),
+		broadcast.track("video").expect("track handle").subscribe(subscription),
+	)
+	.await
+	.expect("subscribe");
+	let mut group = within(&format!("group of {path}"), track.recv_group())
+		.await
+		.expect("recv group")
+		.expect("track closed");
+	within(&format!("frame of {path}"), group.read_frame())
+		.await
+		.expect("read frame")
+		.expect("group closed");
+}
+
+/// Each path `origin` routes now, with its route's hop ids, oldest first.
+///
+/// Read from a fresh cursor's initial set, which reflects the route table as it
+/// stands, unlike a live cursor that holds a best-route change back briefly.
+async fn routes(origin: &origin::Producer) -> BTreeMap<String, Vec<u64>> {
+	let mut announced = origin.consume().announced();
+	let mut routes = BTreeMap::new();
+	loop {
+		match announced.next().await.expect("origin closed") {
+			announce::Event::Start(announce) | announce::Event::Update(announce) => {
+				let hops = announce.route.hops.iter().map(|hop| hop.id()).collect();
+				routes.insert(announce.prefix.as_str().to_string(), hops);
+			}
+			announce::Event::End(announce) => {
+				routes.remove(announce.prefix.as_str());
+			}
+			announce::Event::Live => return routes,
+		}
+	}
+}
+
+/// Wait until `origin`'s routes satisfy `ready`, and return them.
+async fn routes_when(
+	step: &str,
+	origin: &origin::Producer,
+	ready: impl Fn(&BTreeMap<String, Vec<u64>>) -> bool,
+) -> BTreeMap<String, Vec<u64>> {
+	within(step, async {
+		// Created first, so a change landing between two reads still wakes us.
+		let mut changes = origin.consume().announced();
+		loop {
+			let routes = routes(origin).await;
+			if ready(&routes) {
+				return routes;
+			}
+			changes.next().await.expect("origin closed");
+		}
+	})
+	.await
+}
+
+fn has(routes: &BTreeMap<String, Vec<u64>>, paths: &[&str]) -> bool {
+	paths.iter().all(|path| routes.contains_key(*path))
+}
+
+const R: u64 = 1;
+const C1: u64 = 11;
+const C2: u64 = 12;
+const E1: u64 = 21;
+const E2: u64 = 22;
+
+/// Two regions: two edges whose links to two cores are upstream, and one relay
+/// that also serves clients. Both edges send a path to the same core, an edge's
+/// broadcast reaches the far relay, and losing a core moves only its paths.
+#[test]
+fn tiered_regions() {
+	run_cluster_test(tiered_regions_inner());
+}
+
+async fn tiered_regions_inner() {
+	let r = node(R, vec![]).await;
+	let c1 = node(C1, vec![r.dial(false, false)]).await;
+	let c2 = node(C2, vec![r.dial(false, false)]).await;
+	let e1 = node(E1, vec![c1.dial(true, false), c2.dial(true, false)]).await;
+	let e2 = node(E2, vec![c1.dial(true, false), c2.dial(true, false)]).await;
+
+	let paths: Vec<String> = (0..8).map(|i| format!("live/{i}")).collect();
+	let paths: Vec<&str> = paths.iter().map(String::as_str).collect();
+	let _published: Vec<_> = paths.iter().map(|path| publish(&r.origin, path)).collect();
+	routes_when("both cores learn the far region", &c1.origin, |routes| has(routes, &paths)).await;
+	routes_when("both cores learn the far region", &c2.origin, |routes| has(routes, &paths)).await;
+
+	// A route is delivered before a later one: once an edge holds both cores'
+	// markers, it holds both cores' routes to the far region too.
+	let _c1_mark = publish(&c1.origin, "zz/c1");
+	let _c2_mark = publish(&c2.origin, "zz/c2");
+	let marks = ["zz/c1", "zz/c2"];
+	let at_e1 = routes_when("edge 1 hears both cores", &e1.origin, |routes| has(routes, &marks)).await;
+	let at_e2 = routes_when("edge 2 hears both cores", &e2.origin, |routes| has(routes, &marks)).await;
+
+	// Every edge sends a given path to the same core, so the far region serves
+	// one subscription per broadcast whichever edges watch it.
+	let core = |routes: &BTreeMap<String, Vec<u64>>, path: &str| *routes[path].last().expect("hops");
+	for path in &paths {
+		assert_eq!(at_e1[*path].len(), 2, "{path} at edge 1: {:?}", at_e1[*path]);
+		assert_eq!(core(&at_e1, path), core(&at_e2, path), "{path} split across cores");
+	}
+	read_frame(&e1.origin, paths[0]).await;
+	read_frame(&e2.origin, paths[0]).await;
+
+	// A client on the far relay reaches a broadcast published on an edge.
+	let _edge = publish(&e1.origin, "edge/cam");
+	read_frame(&r.origin, "edge/cam").await;
+
+	// Losing a core moves only its paths.
+	let on = |core_id: u64| -> Vec<&str> { paths.iter().copied().filter(|p| core(&at_e1, p) == core_id).collect() };
+	let (on_c1, on_c2) = (on(C1), on(C2));
+	assert!(!on_c1.is_empty() && !on_c2.is_empty(), "paths all on one core: {at_e1:?}");
+	drop(c1);
+	let after = routes_when("edge 1 drops the lost core", &e1.origin, |routes| {
+		has(routes, &paths) && routes.values().all(|hops| !hops.contains(&C1))
+	})
+	.await;
+	for path in &on_c2 {
+		assert_eq!(after[*path], at_e1[*path], "{path} moved off the surviving core");
+	}
+	for path in &on_c1 {
+		assert_eq!(after[*path], [R, C2], "{path} did not fail over");
+	}
+}
+
+/// A route from one core never reaches the other through an edge, whether the
+/// edge dialed the core or the core dialed the edge.
+#[test]
+fn edges_never_transit_between_cores() {
+	run_cluster_test(edges_never_transit_between_cores_inner());
+}
+
+async fn edges_never_transit_between_cores_inner() {
+	let c1 = node(C1, vec![]).await;
+	let e1 = node(E1, vec![c1.dial(true, false)]).await;
+	// The core dials this edge, which still treats it as upstream.
+	let c2 = node(C2, vec![e1.dial(false, true)]).await;
+	let e2 = node(E2, vec![c1.dial(true, false), c2.dial(true, false)]).await;
+
+	let _core = publish(&c1.origin, "live/c1");
+	routes_when("edge 1 learns core 1", &e1.origin, |routes| has(routes, &["live/c1"])).await;
+	routes_when("edge 2 learns core 1", &e2.origin, |routes| has(routes, &["live/c1"])).await;
+
+	// Each edge's own broadcast reaches core 2 after anything it would offer from
+	// core 1, so once both arrive, core 2 has seen every offer it will get.
+	let _e1 = publish(&e1.origin, "zz/e1");
+	let _e2 = publish(&e2.origin, "zz/e2");
+	let at_c2 = routes_when("core 2 hears both edges", &c2.origin, |routes| {
+		has(routes, &["zz/e1", "zz/e2"])
+	})
+	.await;
+	assert!(
+		!at_c2.contains_key("live/c1"),
+		"core 1's broadcast reached core 2 through an edge: {at_c2:?}"
+	);
+	read_frame(&c2.origin, "zz/e1").await;
+	read_frame(&c1.origin, "zz/e2").await;
+}
+
+/// A drone mesh node, a drone with a CDN uplink marked upstream, and a CDN relay
+/// in a line: the mesh reaches the CDN and the CDN reaches the mesh.
+#[test]
+fn uplink_transits_both_ways() {
+	run_cluster_test(uplink_transits_both_ways_inner());
+}
+
+async fn uplink_transits_both_ways_inner() {
+	let cdn = node(31, vec![]).await;
+	let uplink = node(32, vec![cdn.dial(true, false)]).await;
+	let mesh = node(33, vec![uplink.dial(false, false)]).await;
+
+	let _mesh = publish(&mesh.origin, "mesh/cam");
+	let _cdn = publish(&cdn.origin, "cdn/cam");
+	read_frame(&cdn.origin, "mesh/cam").await;
+	read_frame(&mesh.origin, "cdn/cam").await;
+}

From 6d1590a58a11438a30f3535501b06171615fc8c0 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 4 Oct 2026 21:01:22 -0700
Subject: [PATCH 4/8] docs(relay): document upstream links and the upstream
 grant claim

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 doc/bin/relay/auth.md       |  7 ++++---
 doc/bin/relay/cluster.md    | 34 +++++++++++++++++++++++++++++++++-
 doc/bin/relay/config.md     |  2 +-
 doc/lib/rs/moq-auth.md      |  2 +-
 doc/lib/rs/moq-net.md       |  2 +-
 js/auth/src/contract.ts     |  5 +++++
 js/auth/src/interop.test.ts |  3 ++-
 7 files changed, 47 insertions(+), 8 deletions(-)

diff --git a/doc/bin/relay/auth.md b/doc/bin/relay/auth.md
index 5b624bc874..8803e863ff 100644
--- a/doc/bin/relay/auth.md
+++ b/doc/bin/relay/auth.md
@@ -52,9 +52,10 @@ refuses the grant),
 `expires`
 (optional unix seconds; the session closes then), `revalidate` (optional
 seconds until the relay asks again), `tier` (optional label handed to
-[stats](/bin/relay/config#stats)), and `peer` (optional; `true` marks another
+[stats](/bin/relay/config#stats)), `peer` (optional; `true` marks another
 relay, so what it announces counts as entering the cluster elsewhere, not as
-ingest here). A 2xx with a grant admits. A 401 or 403 refuses.
+ingest here), and `upstream` (optional; `true` marks that peer as an
+[upstream link](/bin/relay/cluster#upstream-links), and requires `peer`). A 2xx with a grant admits. A 401 or 403 refuses.
 Anything else at connect, a timeout, a 5xx, or an unparseable body, refuses and
 logs an error; nothing is admitted because the server was down. A grant that
 names nothing refuses, and one with `revalidate` but no `expires` is refused
@@ -66,7 +67,7 @@ clock, so later polls, outages, and wall-clock adjustments do not restart it.
 **Revalidate and outage.** On the cadence the relay POSTs `revalidate` with the
 same request. A grant applies: a changed `root` or `mounts`, or one that no longer covers
 what the session holds closes it with `Unauthorized` (the live session is not
-resized in place), and so does a flipped `peer`; a changed `tier` keeps the
+resized in place), and so does a flipped `peer` or `upstream`; a changed `tier` keeps the
 session and moves its stats: its presence counts under the new tier from then
 on, as does each group and subscription it starts afterwards, while one already
 in flight finishes where it began. A 401 or 403 closes the session now, as does a 2xx
diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md
index 635f7bbfcb..7fa6e10ca4 100644
--- a/doc/bin/relay/cluster.md
+++ b/doc/bin/relay/cluster.md
@@ -75,6 +75,38 @@ 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.
 
+## Upstream links
+
+Mark a link `upstream` and this relay never offers it a route learned on
+another upstream link. Everything else still transits, and clients still see
+every route the relay knows. An edge marks its region's cores upstream, so it
+reaches every core and every core reaches its clients, but it never carries
+traffic from one core to another. A drone marks its CDN uplink upstream, so its
+mesh reaches the CDN and the CDN reaches the mesh, but the drone never carries
+traffic between CDN relays.
+
+```toml
+# edge.toml
+[cluster]
+connect = [
+  { url = "https://core-a.example.com/", upstream = true },
+  { url = "https://core-b.example.com/", upstream = true },
+]
+```
+
+The mark belongs to the link, whichever side dialed. A peer this relay dials
+is upstream when its `connect` or `connect_api` entry says so. A peer that
+dials in is upstream when its [grant](/bin/relay/auth#the-contract) sets
+`"upstream": true` beside `"peer": true`, so an edge whose auth server grants
+that to core certificates treats a core that dials it as upstream too. A
+relay that predates the mark treats every link as transit, so a cluster
+migrates one region at a time.
+
+Which relay dials which is still the peer list's job: there are no roles and
+no topology check, so whatever generates the list applies the layout's rules.
+Hiding a core from clients is admission, not routing: its auth refuses any
+session that is not a cluster peer.
+
 ## Link costs
 
 Add `?cost=N` to a peer URL to route by price instead of hop count. An unpriced
@@ -101,7 +133,7 @@ connect = ["https://sibling.same-dc/?cost=0", "https://us-east.example.com/?cost
 ```
 
 The same policy reads as an object, which is the only form that accepts
-`egress` and `token`. A bare URL stays valid, and an object whose `url` still
+`egress`, `token`, and [`upstream`](#upstream-links). A bare URL stays valid, and an object whose `url` still
 carries `?cost=` or `?jwt=` alongside those fields is rejected rather than
 given a precedence a migration could silently get wrong.
 
diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md
index 0524a10e71..ba077f96d2 100644
--- a/doc/bin/relay/config.md
+++ b/doc/bin/relay/config.md
@@ -151,7 +151,7 @@ See [Authentication](/bin/relay/auth).
 
 ```toml
 [cluster]
-connect = ["https://us-east.example.com/?cost=10"]   # Peers to dial. ?cost prices the link, or use {url, cost, egress, token} objects.
+connect = ["https://us-east.example.com/?cost=10"]   # Peers to dial. ?cost prices the link, or use {url, cost, egress, token, upstream} objects.
 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=.
diff --git a/doc/lib/rs/moq-auth.md b/doc/lib/rs/moq-auth.md
index 6375356ca5..5060f15d3b 100644
--- a/doc/lib/rs/moq-auth.md
+++ b/doc/lib/rs/moq-auth.md
@@ -13,7 +13,7 @@ Everything a party needs to ask for or answer an authorization on
 answers the relay, in a service that mints tokens for clients, or in your own
 accept loop that decides in process.
 
-- **Request and grant**: `Request` is the JSON a relay POSTs per session event (`connect`, `revalidate`, `end`) with everything it knows: id, node, transport, addresses, SNI and ALPN, the raw path and query, the moq-transport SETUP `token` (its Token Type and bytes, base64url on the wire), the declared role, and the verified certificate facts. `Grant` is the answer: `publish` and `subscribe` pattern unions, an optional `root` alias, `mounts` that read a subtree from elsewhere on the origin, `expires`, `revalidate`, `tier`, and `peer`, which marks the session as a cluster peer so the routes it announces report `Source::Peer`. `Grant::validate` refuses a grant that names nothing, asks to be revalidated without a bound or at no interval, or has already expired; expiry is exact, with no grace for clock skew.
+- **Request and grant**: `Request` is the JSON a relay POSTs per session event (`connect`, `revalidate`, `end`) with everything it knows: id, node, transport, addresses, SNI and ALPN, the raw path and query, the moq-transport SETUP `token` (its Token Type and bytes, base64url on the wire), the declared role, and the verified certificate facts. `Grant` is the answer: `publish` and `subscribe` pattern unions, an optional `root` alias, `mounts` that read a subtree from elsewhere on the origin, `expires`, `revalidate`, `tier`, `peer`, which marks the session as a cluster peer so the routes it announces report `Source::Peer`, and `upstream`, which marks that peer an [upstream link](/bin/relay/cluster#upstream-links). `Grant::validate` refuses a grant that names nothing, marks a non-peer upstream, asks to be revalidated without a bound or at no interval, or has already expired; expiry is exact, with no grace for clock skew.
 - **Lease**: `lease::Producer` and `lease::Consumer` are the handle a session holds for its grant. The consumer reads the current grant, waits for a change, and learns why the lease ended; the producer updates and revokes. Either side's terminal call returns the reason the lease actually ended with, so whichever got there first is what both report. `Consumer::fixed` is a grant nobody drives. `Consumer::revalidate` nudges a re-check now and the producer observes it via `poll_revalidate` or `revalidate_requested`, which is how the relay's session push lands. Whoever runs the accept loop builds the producer, so an embedder decides in process with no trait and no HTTP. Enforcing `expires` is the holder's job; the `Client` driver also revokes at expiry so its `end` event goes out. `Grant::deadline()` (feature `tokio`, also enabled by `client` and `serve`) snapshots expiry on Tokio's clock. Call it once per accepted grant and retain the deadline; an expiry already past is now.
 - **Client**: `Client::new(url, tls)` and `Client::connect(request)` drive a lease against an auth server over `https://`, `unix://`, or loopback `http://`: revalidate on cadence with jittered backoff through an outage until `expires`, revoke on a 401/403 or an invalid grant, and POST `end` with the reason, duration, and byte totals the session reported through `lease::Consumer::close` when it ended. Dropping the consumer reports zero bytes. `end.reason` is `dropped`, `expired`, `refused`, `invalid`, `narrowed`, `shutdown`, or the session's own classification.
 - **Server**: `serve::Policy` and `serve::Server` (feature `serve`) are the reference auth server behind `moq auth serve`: a JWT from the `jwt` query or a type-0 SETUP token, verified against a key file, a `{kid}.jwk` directory, or a JWK Set file (a token of another type, or two different ones, is refused), an explicit grant for verified certificates, the anonymous permissions, a tier, the revalidation cadence, a default `expires`, and live session caps per token and per remote address. A token is authorized at the dialed path with `Claims::authorize`; residuals become the grant. `Server::router` is an axum `POST /` you can mount in your own service.
diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md
index 6efaecabab..fe2ba56a37 100644
--- a/doc/lib/rs/moq-net.md
+++ b/doc/lib/rs/moq-net.md
@@ -23,7 +23,7 @@ above ([hang](/lib/rs/hang)); relays and CDNs implement only this.
 - **Groups** are written frame by frame and delivered on independent streams. Old groups are cached for fetch-by-sequence; stale groups are skipped per the subscriber's budget.
 - **Track ends**: `finish()` ends a track at its live edge, while `finish_at(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber awaits it with `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF.
 - **Datagrams** send a single small frame unreliably on moq-lite 05+ and moq-transport.
-- **Routes** record the relay hops and a cost, which is what the relay [cluster](/bin/relay/cluster) routes on. A hop of 0 marks the chain anonymous: `Route::is_anonymous()` is true, and that route ranks below every fully identified one. `Route::source()` says where a delivered route entered: `Source::Local`, or `Source::Peer(hop)` when a handle marked `origin::Producer::peer()` announced it. `origin::Consumer::local()` sees only the local ones.
+- **Routes** record the relay hops and a cost, which is what the relay [cluster](/bin/relay/cluster) routes on. A hop of 0 marks the chain anonymous: `Route::is_anonymous()` is true, and that route ranks below every fully identified one. `Route::source()` says where a delivered route entered: `Source::Local`, or `Source::Peer(hop)` when a handle marked `origin::Producer::peer()` announced it. `origin::Consumer::local()` sees only the local ones. A handle marked `origin::Producer::upstream()` announces as a peer, and a consumer taken from it never sees a route another upstream handle announced, so a relay never transits between two upstream links.
 - **Stats** counters per broadcast and session, drained by [`moq-stats`](https://docs.rs/moq-stats).
 
 It runs over anything implementing `web_transport_trait::poll::Session`: noq, the
diff --git a/js/auth/src/contract.ts b/js/auth/src/contract.ts
index 19b31a9603..39e4ff238e 100644
--- a/js/auth/src/contract.ts
+++ b/js/auth/src/contract.ts
@@ -137,6 +137,8 @@ export const GrantSchema = z
 		tier: z.optional(z.string()),
 		/** The session is a cluster peer (another relay): what it announces entered the cluster elsewhere. */
 		peer: z.optional(z.boolean()),
+		/** The peer is upstream: the relay never offers it a route learned from another upstream. Requires `peer`. */
+		upstream: z.optional(z.boolean()),
 	})
 	.check(
 		z.refine((grant) => (grant.publish?.length ?? 0) > 0 || (grant.subscribe?.length ?? 0) > 0, {
@@ -145,5 +147,8 @@ export const GrantSchema = z
 		z.refine((grant) => grant.revalidate === undefined || grant.expires !== undefined, {
 			message: "a grant that asks to be revalidated must expire",
 		}),
+		z.refine((grant) => !grant.upstream || grant.peer === true, {
+			message: "an upstream grant must be a peer",
+		}),
 	);
 export type Grant = z.infer<typeof GrantSchema>;
diff --git a/js/auth/src/interop.test.ts b/js/auth/src/interop.test.ts
index 85c10ad522..bd2b4b3d22 100644
--- a/js/auth/src/interop.test.ts
+++ b/js/auth/src/interop.test.ts
@@ -261,7 +261,7 @@ describe("Rust-serialized contract", () => {
 	const RUST_INVALID_END =
 		'{"id":"00ff","event":"end","reason":"invalid","duration":1.5,"bytes":{"sent":10,"received":20},"node":"relay-1","transport":"websocket","remote":"203.0.113.9:4433","path":"/demo/room","query":"jwt=abc"}';
 	const RUST_GRANT =
-		'{"publish":["alice/**"],"subscribe":["**"],"root":"pid/room","expires":4102444800,"revalidate":60,"tier":"websocket","peer":true}';
+		'{"publish":["alice/**"],"subscribe":["**"],"root":"pid/room","expires":4102444800,"revalidate":60,"tier":"websocket","peer":true,"upstream":true}';
 
 	test("parse a request a Rust relay sent", () => {
 		const request = RequestSchema.parse(JSON.parse(RUST_REQUEST));
@@ -298,6 +298,7 @@ describe("Rust-serialized contract", () => {
 			revalidate: 60,
 			tier: "websocket",
 			peer: true,
+			upstream: true,
 		});
 		expect(JSON.stringify(grant)).toBe(RUST_GRANT);
 	});

From 422e9b78f562e02bd9a5948f1b5880850c351a03 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 4 Oct 2026 21:08:58 -0700
Subject: [PATCH 5/8] feat(tokio): qmux over TLS on TCP via tls://

A tls:// URL dials qmux over TLS on TCP with the connect TLS settings,
and listen.tcp.tls serves the TCP listener over TLS with the listen
certificate. The listener asks for no client certificate, since qmux's
TLS accept keeps no peer identity for the auth server.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 doc/bin/relay/auth.md                |  11 +-
 doc/bin/relay/cluster.md             |  20 ++++
 doc/bin/relay/config.md              |   1 +
 rs/moq-auth/src/request.rs           |   2 +-
 rs/moq-relay/tests/upstream_links.rs |  11 +-
 rs/moq-tokio/Cargo.toml              |   2 +-
 rs/moq-tokio/src/client.rs           |  30 ++++-
 rs/moq-tokio/src/connect.rs          |   2 +-
 rs/moq-tokio/src/connection.rs       |   8 +-
 rs/moq-tokio/src/lib.rs              |   2 +-
 rs/moq-tokio/src/listen.rs           |   4 +
 rs/moq-tokio/src/server.rs           |  86 ++++++++++++--
 rs/moq-tokio/src/settings.rs         |   3 +
 rs/moq-tokio/src/tcp.rs              | 168 +++++++++++++++++++++++----
 14 files changed, 303 insertions(+), 47 deletions(-)

diff --git a/doc/bin/relay/auth.md b/doc/bin/relay/auth.md
index 8803e863ff..9fe0373f35 100644
--- a/doc/bin/relay/auth.md
+++ b/doc/bin/relay/auth.md
@@ -385,7 +385,12 @@ bind = "/run/moq/internal.sock"
 allow.uid = [1001]
 ```
 
-Bind TCP to loopback or a private interface; it carries no peer identity. The
-Unix socket is created mode `0666`, so gate it with a restrictive parent
-directory or an explicit allowlist.
+Bind plaintext TCP to loopback or a private interface; it carries no peer
+identity. The Unix socket is created mode `0666`, so gate it with a restrictive
+parent directory or an explicit allowlist.
 These are native-only paths for gateways and stats publishers on the same host.
+
+`listen.tcp.tls = true` serves the TCP listener over TLS with the listen
+certificate instead, for `tls://` dials such as [cluster links](/bin/relay/cluster#tls-links)
+that need neither QUIC nor a WebSocket. It still carries no peer identity: it
+asks for no client certificate, so a peer on it presents a token, not mTLS.
diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md
index 7fa6e10ca4..75d1cd1bee 100644
--- a/doc/bin/relay/cluster.md
+++ b/doc/bin/relay/cluster.md
@@ -107,6 +107,26 @@ no topology check, so whatever generates the list applies the layout's rules.
 Hiding a core from clients is admission, not routing: its auth refuses any
 session that is not a cluster peer.
 
+## TLS links
+
+A link inside a region is rarely congested, so it can skip QUIC: a `tls://`
+peer URL dials qmux over TLS on TCP, verified with the same `connect.tls`
+settings as any other dial. The accepting relay serves it from its TCP
+listener with TLS on. That listener asks for no client certificate, so a peer
+on it authenticates with a token (`cluster.token`, `token`, or `?jwt=`), not
+mTLS.
+
+```toml
+# core.toml
+[listen.tcp]
+bind = "[::]:4443"
+tls = true
+
+# edge.toml
+[cluster]
+connect = [{ url = "tls://core-a.internal:4443/", upstream = true }]
+```
+
 ## Link costs
 
 Add `?cost=N` to a peer URL to route by price instead of hop count. An unpriced
diff --git a/doc/bin/relay/config.md b/doc/bin/relay/config.md
index ba077f96d2..b224436daa 100644
--- a/doc/bin/relay/config.md
+++ b/doc/bin/relay/config.md
@@ -29,6 +29,7 @@ root = ["peer-ca.pem"]               # Optional: CAs for client certs (mTLS), re
 
 [listen.tcp]                         # Plaintext qmux over TCP for trusted local workers.
 bind = "127.0.0.1:4444"
+# tls = true                         # Or: qmux over TLS (tls://) with the listen certificate, no client certs.
 
 [listen.unix]                        # Plaintext qmux over a Unix socket, gated by peer credentials.
 bind = "/run/moq/internal.sock"
diff --git a/rs/moq-auth/src/request.rs b/rs/moq-auth/src/request.rs
index 2e7d20504e..1c5bcfd78c 100644
--- a/rs/moq-auth/src/request.rs
+++ b/rs/moq-auth/src/request.rs
@@ -135,7 +135,7 @@ pub enum Transport {
 	Iroh,
 	/// A WebSocket connection using qmux framing.
 	WebSocket,
-	/// A plaintext TCP connection using qmux framing.
+	/// A TCP connection using qmux framing, plaintext or TLS.
 	Tcp,
 	/// A Unix domain socket using qmux framing.
 	Unix,
diff --git a/rs/moq-relay/tests/upstream_links.rs b/rs/moq-relay/tests/upstream_links.rs
index b1333f4d55..03f3fa8074 100644
--- a/rs/moq-relay/tests/upstream_links.rs
+++ b/rs/moq-relay/tests/upstream_links.rs
@@ -2,9 +2,10 @@
 //! one upstream link to another, whichever side dialed, and everything else
 //! still transits.
 //!
-//! Each relay is a [`cluster::Cluster`] serving accepted sessions through an
-//! embedded auth decider that reads the dialed path: `/upstream` admits the
-//! dialer as an upstream peer, `/peer` as a plain one.
+//! Each relay is a [`cluster::Cluster`] serving qmux over TLS on TCP (`tls://`),
+//! the way edge links run, through an embedded auth decider that reads the
+//! dialed path: `/upstream` admits the dialer as an upstream peer, `/peer` as a
+//! plain one.
 
 use std::collections::BTreeMap;
 use std::time::Duration;
@@ -59,7 +60,7 @@ impl Node {
 	/// asks this node to admit the dialer as its upstream.
 	fn dial(&self, upstream: bool, as_upstream: bool) -> Peer {
 		let path = if as_upstream { "upstream" } else { "peer" };
-		Peer::new(format!("tcp://127.0.0.1:{}/{path}", self.port)).with_upstream(upstream)
+		Peer::new(format!("tls://127.0.0.1:{}/{path}", self.port)).with_upstream(upstream)
 	}
 }
 
@@ -80,6 +81,8 @@ async fn node(id: u64, connect: Vec<Peer>) -> Node {
 
 	let mut listen = moq_tokio::listen::Config::default();
 	listen.tcp.bind = Some("127.0.0.1:0".parse().expect("parse addr"));
+	listen.tcp.tls = Some(true);
+	listen.tls.generate = vec!["localhost".into()];
 	let mut listener = listen
 		.init(Default::default())
 		.expect("server init")
diff --git a/rs/moq-tokio/Cargo.toml b/rs/moq-tokio/Cargo.toml
index bbe9c0defa..88b6f65498 100644
--- a/rs/moq-tokio/Cargo.toml
+++ b/rs/moq-tokio/Cargo.toml
@@ -39,7 +39,7 @@ iroh = ["_transport", "dep:web-transport-iroh", "dep:web-transport-proto", "dep:
 mdns = ["dep:mdns-sd", "dep:hmac", "dep:sha2", "dep:subtle"]
 jemalloc = ["dep:tikv-jemallocator", "dep:tikv-jemalloc-ctl"]
 websocket = ["_transport", "dep:qmux", "watch", "dep:dns-lookup"]
-# Plain-TCP qmux transport (`tcp://`), no TLS. Server-side and client-side.
+# Qmux over TCP: plaintext `tcp://`, and `tls://` once a crypto provider is on. Server-side and client-side.
 tcp = ["_transport", "dep:qmux", "dep:dns-lookup"]
 # Unix-domain-socket qmux transport (`unix://`), unix-only. Adds peer-credential
 # inspection on accept. Pulls in `tcp` (shares qmux's stream transport).
diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs
index d7c517d258..de3b156421 100644
--- a/rs/moq-tokio/src/client.rs
+++ b/rs/moq-tokio/src/client.rs
@@ -65,12 +65,16 @@ pub struct Client {
 	/// Whether the TLS config pins a certificate fingerprint, which only verifies
 	/// the configured host, so a GOAWAY may not redirect elsewhere.
 	pub(crate) pinned: bool,
-	/// The resolved Happy Eyeballs timings, used by the `tcp://` dial here; the
-	/// QUIC backend captures its own copy from the config.
+	/// The resolved Happy Eyeballs timings, used by the `tcp://` and `tls://`
+	/// dials here; the QUIC backend captures its own copy from the config.
 	#[cfg(feature = "tcp")]
 	failover_delay: std::time::Duration,
 	#[cfg(feature = "tcp")]
 	resolution_delay: std::time::Duration,
+	/// The TLS settings a `tls://` dial builds from when it runs, so a client
+	/// that only dials plaintext `tcp://` never needs a crypto provider.
+	#[cfg(feature = "tcp")]
+	tcp_tls: crate::tls::Connect,
 	#[cfg(feature = "websocket")]
 	websocket: crate::websocket::Config,
 	/// The TLS server name override used by the WebSocket fallback.
@@ -130,6 +134,8 @@ impl Client {
 		let failover_delay = resolved.race;
 		#[cfg(feature = "tcp")]
 		let resolution_delay = resolved.resolution_delay;
+		#[cfg(feature = "tcp")]
+		let tcp_tls = config.tls.clone();
 		#[cfg(feature = "websocket")]
 		let tls_host_name = config.tls.host_name.clone();
 		let timeout = resolved.timeout;
@@ -154,6 +160,8 @@ impl Client {
 			failover_delay,
 			#[cfg(feature = "tcp")]
 			resolution_delay,
+			#[cfg(feature = "tcp")]
+			tcp_tls,
 			#[cfg(feature = "websocket")]
 			websocket: config.websocket,
 			#[cfg(feature = "websocket")]
@@ -388,6 +396,20 @@ impl Client {
 			return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?);
 		}
 
+		// Qmux over TLS over TCP, for links that need neither QUIC nor a WebSocket.
+		#[cfg(feature = "tcp")]
+		if url.scheme() == "tls" {
+			let session = crate::tcp::connect_tls(
+				url,
+				&self.versions.alpns(),
+				&self.tcp_tls,
+				self.failover_delay,
+				self.resolution_delay,
+			)
+			.await?;
+			return Ok(connect_session(&moq, crate::transport::Session::new(session)).await?);
+		}
+
 		// Unix domain socket (qmux, no TLS). Same-host only; the server can
 		// authenticate us by uid/gid via SO_PEERCRED.
 		#[cfg(all(feature = "uds", unix))]
@@ -551,7 +573,7 @@ fn setup_path(url: &Url) -> Option<String> {
 			.filter(|path| !path.is_empty()),
 		// Raw QUIC and qmux over TCP negotiate an ALPN and nothing else, so the whole
 		// request target travels in the SETUP.
-		"moqt" | "moql" | "tcp" => request_target(url),
+		"moqt" | "moql" | "tcp" | "tls" => request_target(url),
 		_ => None,
 	}
 }
@@ -789,6 +811,8 @@ mod tests {
 			("tcp://localhost:4443/room", Some("/room")),
 			("tcp://localhost:4443/room?jwt=abc", Some("/room?jwt=abc")),
 			("tcp://localhost:4443", None),
+			("tls://localhost:4443/room?jwt=abc", Some("/room?jwt=abc")),
+			("tls://localhost:4443", None),
 			// Raw QUIC: the URL is ours alone, so the path and query have to ride the
 			// SETUP or the server never sees them.
 			("moqt://relay.example.com/anon", Some("/anon")),
diff --git a/rs/moq-tokio/src/connect.rs b/rs/moq-tokio/src/connect.rs
index 3dbbe6a889..6589087eb8 100644
--- a/rs/moq-tokio/src/connect.rs
+++ b/rs/moq-tokio/src/connect.rs
@@ -507,7 +507,7 @@ pub struct Config {
 	/// When DNS returns multiple addresses, attempts alternate between IPv6 and
 	/// IPv4, each starting this long after the previous one (or immediately when
 	/// it fails), and the first connection to complete wins. `0s` dials every
-	/// address at once. Defaults to 250ms. Applies to the QUIC and `tcp://` dials.
+	/// address at once. Defaults to 250ms. Applies to the QUIC, `tcp://`, and `tls://` dials.
 	///
 	/// This staggers the attempts within one [`crate::Client::connect`]; [`Self::timeout`]
 	/// bounds that call as a whole.
diff --git a/rs/moq-tokio/src/connection.rs b/rs/moq-tokio/src/connection.rs
index 1e9178c819..c3c6a727cb 100644
--- a/rs/moq-tokio/src/connection.rs
+++ b/rs/moq-tokio/src/connection.rs
@@ -272,14 +272,14 @@ impl Redirect {
 /// Whether this scheme's dial installs the Rustls verifier, so a configured
 /// fingerprint actually checked the peer. Plain and non-Rustls transports do not.
 fn fingerprint_pins(scheme: &str) -> bool {
-	matches!(scheme, "https" | "wss" | "moqt" | "moql")
+	matches!(scheme, "https" | "wss" | "moqt" | "moql" | "tls")
 }
 
 /// Rank a scheme so a peer-supplied redirect cannot silently drop encryption.
 /// Unknown schemes rank lowest, so a forgotten classification is refused.
 fn scheme_tier(scheme: &str) -> u8 {
 	match scheme {
-		"https" | "moqt" | "moql" | "wss" | "iroh" => 2,
+		"https" | "moqt" | "moql" | "wss" | "iroh" | "tls" => 2,
 		"tcp" | "ws" | "http" => 1,
 		// `unix` lands here deliberately: local IPC is not an upgrade over a
 		// network transport, it is a different reachability class (see `is_local`).
@@ -1553,7 +1553,7 @@ mod tests {
 	/// forgot to add is refused rather than trusted.
 	#[test]
 	fn scheme_tiers_rank_encrypted_above_plaintext() {
-		for scheme in ["https", "moqt", "moql", "wss", "iroh"] {
+		for scheme in ["https", "moqt", "moql", "wss", "iroh", "tls"] {
 			assert_eq!(scheme_tier(scheme), 2, "{scheme} is encrypted");
 		}
 		for scheme in ["tcp", "ws", "http"] {
@@ -1703,7 +1703,7 @@ mod tests {
 	/// must not inherit the pin.
 	#[test]
 	fn a_fingerprint_pin_only_covers_rustls_schemes() {
-		for scheme in ["https", "wss", "moqt", "moql"] {
+		for scheme in ["https", "wss", "moqt", "moql", "tls"] {
 			assert!(fingerprint_pins(scheme), "{scheme}");
 		}
 		for scheme in ["http", "ws", "tcp", "unix", "iroh"] {
diff --git a/rs/moq-tokio/src/lib.rs b/rs/moq-tokio/src/lib.rs
index f520b86565..e5bea556af 100644
--- a/rs/moq-tokio/src/lib.rs
+++ b/rs/moq-tokio/src/lib.rs
@@ -4,7 +4,7 @@
 //! - WebTransport (HTTP/3)
 //! - Raw QUIC (with ALPN negotiation)
 //! - WebSocket (fallback via [web-transport-ws](https://crates.io/crates/web-transport-ws))
-//! - Plain TCP via the `tcp://` scheme (qmux, no TLS; requires `tcp` feature)
+//! - TCP via the `tcp://` (plaintext) and `tls://` schemes (qmux; requires `tcp` feature)
 //! - Unix domain socket via the `unix://` scheme (qmux, peer-credential aware; requires `uds` feature, unix-only)
 //! - Iroh P2P (requires `iroh` feature)
 //!
diff --git a/rs/moq-tokio/src/listen.rs b/rs/moq-tokio/src/listen.rs
index cbd88f64bd..03610d519b 100644
--- a/rs/moq-tokio/src/listen.rs
+++ b/rs/moq-tokio/src/listen.rs
@@ -423,6 +423,10 @@ impl Config {
 
 	#[cfg(feature = "_transport")]
 	pub(crate) fn validate(&self) -> crate::Result<()> {
+		#[cfg(feature = "tcp")]
+		if self.tcp.tls == Some(true) && self.tcp.bind.is_none() {
+			return Err(crate::Error::NoBackend("--listen-tcp-tls requires --listen-tcp-bind"));
+		}
 		Ok(())
 	}
 
diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs
index 1a360703b8..871f4ea7e6 100644
--- a/rs/moq-tokio/src/server.rs
+++ b/rs/moq-tokio/src/server.rs
@@ -329,7 +329,11 @@ impl Server {
 		let mut stream_binds = Vec::new();
 		#[cfg(feature = "tcp")]
 		if let Some(addr) = config.tcp.bind.filter(|_| build_streams) {
-			stream_binds.push(StreamBind::Tcp(addr));
+			let tls = match config.tcp.tls == Some(true) {
+				true => Some(TcpTls::new(&config, &versions)?),
+				false => None,
+			};
+			stream_binds.push(StreamBind::Tcp(addr, tls));
 		}
 		#[cfg(all(feature = "uds", unix))]
 		if let Some(path) = config.unix.bind.clone().filter(|_| build_streams) {
@@ -798,7 +802,7 @@ impl Listener {
 		self.server.websocket_local_addr()
 	}
 
-	/// The address the plain TCP (qmux) listener bound to, if one was configured.
+	/// The address the TCP (qmux) listener bound to, if one was configured.
 	#[cfg(feature = "tcp")]
 	pub fn tcp_local_addr(&self) -> Option<net::SocketAddr> {
 		self.server.streams.tcp_local_addr
@@ -988,12 +992,66 @@ fn stream_versions(base: &moq_net::Versions) -> moq_net::Versions {
 /// A configured stream listener (`--listen-tcp-bind` / `--listen-unix-bind`).
 #[cfg(any(feature = "tcp", all(feature = "uds", unix)))]
 enum StreamBind {
+	/// With its TLS when the listener serves `tls://`.
 	#[cfg(feature = "tcp")]
-	Tcp(net::SocketAddr),
+	Tcp(net::SocketAddr, Option<TcpTls>),
 	#[cfg(all(feature = "uds", unix))]
 	Unix(PathBuf),
 }
 
+/// The TLS a `--listen-tcp-tls` listener serves, reloading its certificate
+/// while this lives.
+#[cfg(feature = "tcp")]
+struct TcpTls {
+	config: std::sync::Arc<rustls::ServerConfig>,
+	#[cfg(all(feature = "watch", feature = "_certs"))]
+	_reload: crate::tls::ReloadingServerConfig,
+}
+
+#[cfg(all(feature = "tcp", feature = "_certs"))]
+impl TcpTls {
+	/// Serve the listen certificate, accepting each stream version as a
+	/// `qmux-01.<alpn>` TLS ALPN.
+	///
+	/// Asks for no client certificate: qmux's TLS accept keeps no peer identity
+	/// for the auth server to read, so a `tls.root` meant for QUIC peers does not
+	/// apply, and a peer on this listener authenticates with a token.
+	fn new(config: &crate::listen::Config, versions: &moq_net::Versions) -> crate::Result<Self> {
+		let mut listen = config.tls.clone();
+		listen.root.clear();
+		let alpn = stream_versions(versions)
+			.alpns()
+			.iter()
+			.map(|alpn| format!("{}{alpn}", qmux::Version::QMux01.prefix()).into_bytes())
+			.collect();
+		Self::serve(&listen, alpn)
+	}
+
+	#[cfg(feature = "watch")]
+	fn serve(listen: &crate::tls::Listen, alpn: Vec<Vec<u8>>) -> crate::Result<Self> {
+		let reload = listen.server_config_reloading(alpn)?;
+		Ok(Self {
+			config: reload.config(),
+			_reload: reload,
+		})
+	}
+
+	#[cfg(not(feature = "watch"))]
+	fn serve(listen: &crate::tls::Listen, alpn: Vec<Vec<u8>>) -> crate::Result<Self> {
+		Ok(Self {
+			config: listen.server_config(alpn)?,
+		})
+	}
+}
+
+#[cfg(all(feature = "tcp", not(feature = "_certs")))]
+impl TcpTls {
+	/// Without a crypto provider there is no certificate to serve.
+	fn new(_config: &crate::listen::Config, _versions: &moq_net::Versions) -> crate::Result<Self> {
+		Err(Error::NoBackend("--listen-tcp-tls requires a crypto provider feature"))
+	}
+}
+
 /// A bound stream listener, before its accept loop is spawned.
 #[cfg(any(feature = "tcp", all(feature = "uds", unix)))]
 enum BoundListener {
@@ -1009,7 +1067,7 @@ impl StreamBind {
 	fn name(&self) -> &'static str {
 		match self {
 			#[cfg(feature = "tcp")]
-			Self::Tcp(_) => "tcp",
+			Self::Tcp(..) => "tcp",
 			#[cfg(all(feature = "uds", unix))]
 			Self::Unix(_) => "unix",
 		}
@@ -1025,6 +1083,9 @@ impl StreamBind {
 /// stopped when the [`Listener`] closes or drops, so no socket lingers.
 #[cfg(any(feature = "tcp", all(feature = "uds", unix)))]
 struct StreamListeners {
+	/// Keeps the TCP listener's TLS certificate reloading while it serves.
+	#[cfg(feature = "tcp")]
+	_tcp_tls: Option<TcpTls>,
 	binds: Vec<StreamBind>,
 	/// One per entry in `binds`, in the same order, and created up front rather than
 	/// with the listener: an owner registering these with a metrics endpoint does so
@@ -1057,6 +1118,8 @@ impl StreamListeners {
 			.map(|bind| crate::accept::Health::new(bind.name()))
 			.collect();
 		Self {
+			#[cfg(feature = "tcp")]
+			_tcp_tls: None,
 			binds,
 			health,
 			versions,
@@ -1086,16 +1149,21 @@ impl StreamListeners {
 			let alpns = self.versions.alpns();
 			match bind {
 				#[cfg(feature = "tcp")]
-				StreamBind::Tcp(addr) => {
-					if !addr.ip().is_loopback() {
+				StreamBind::Tcp(addr, tls) => {
+					if tls.is_none() && !addr.ip().is_loopback() {
 						tracing::warn!(%addr, "tcp listener bound to a non-loopback address; qmux is UNENCRYPTED, ensure the network is trusted");
 					}
-					let listener = crate::tcp::Listener::bind(addr)
+					let mut listener = crate::tcp::Listener::bind(addr)
 						.await?
 						.with_protocols(alpns)
 						.with_accept_health(health);
+					let encrypted = tls.is_some();
+					if let Some(tls) = tls {
+						listener = listener.with_tls(tls.config.clone());
+						self._tcp_tls = Some(tls);
+					}
 					let local = listener.local_addr()?;
-					tracing::info!(addr = %local, "listening (tcp)");
+					tracing::info!(addr = %local, tls = encrypted, "listening (tcp)");
 					self.tcp_local_addr = Some(local);
 					pending.push(BoundListener::Tcp(listener));
 				}
@@ -1338,7 +1406,7 @@ pub enum Transport {
 	Iroh,
 	/// A WebSocket connection using qmux framing.
 	WebSocket,
-	/// A plaintext TCP connection using qmux framing.
+	/// A TCP connection using qmux framing, plaintext or TLS.
 	Tcp,
 	/// A Unix domain socket using qmux framing.
 	Unix,
diff --git a/rs/moq-tokio/src/settings.rs b/rs/moq-tokio/src/settings.rs
index bec2ac4d22..25a81911fc 100644
--- a/rs/moq-tokio/src/settings.rs
+++ b/rs/moq-tokio/src/settings.rs
@@ -72,6 +72,9 @@ struct ListenTls {
 struct ListenTcp {
 	#[usage(env = "MOQ_LISTEN_TCP_BIND", cli("--listen-tcp-bind"))]
 	bind: Option<String>,
+
+	#[usage(env = "MOQ_LISTEN_TCP_TLS", cli("--listen-tcp-tls"))]
+	tls: Option<bool>,
 }
 
 /// `[listen.unix]` / `--listen-unix-*`.
diff --git a/rs/moq-tokio/src/tcp.rs b/rs/moq-tokio/src/tcp.rs
index b275388633..09d4db8653 100644
--- a/rs/moq-tokio/src/tcp.rs
+++ b/rs/moq-tokio/src/tcp.rs
@@ -1,11 +1,13 @@
-//! Plain-TCP qmux transport, reachable via the `tcp://` URL scheme.
+//! Qmux over TCP: plaintext via the `tcp://` URL scheme, TLS via `tls://`.
 //!
-//! Runs the QMux wire format directly over TCP with no TLS or WebSocket
-//! framing. There is no transport encryption and no authentication, so only
-//! use this on a trusted network (loopback, a private VPC interface, etc.).
+//! Both run the QMux wire format directly over TCP with no WebSocket framing.
+//! Plaintext `tcp://` has no transport encryption and no authentication, so only
+//! use it on a trusted network (loopback, a private VPC interface, etc.).
+//! `tls://` encrypts and verifies the server with the same TLS settings as the
+//! other transports, for links that need neither QUIC nor a WebSocket.
 //!
-//! TCP has no TLS handshake, so the application protocol (the moq ALPN) is
-//! negotiated in-band: pass the offered/supported protocols and the resulting
+//! Plaintext TCP has no TLS handshake, so the application protocol (the moq
+//! ALPN) is negotiated in-band; TLS negotiates it in its handshake. Either way
 //! `qmux::Session::protocol()` is populated before connect/accept returns.
 
 use std::net;
@@ -15,11 +17,12 @@ use url::Url;
 /// negotiated) since there's no TLS ALPN to carry it.
 const WIRE_VERSION: qmux::Version = qmux::Version::QMux01;
 
-/// Plaintext-TCP qmux listener settings (no TLS, no UDP).
+/// Qmux TCP listener settings, plaintext or TLS (no UDP).
 ///
-/// Flattened onto [`crate::listen::Config::tcp`]. TCP carries no peer identity, so
-/// the listener must only be reachable from trusted clients. Bind it to loopback
-/// or a private interface; a non-loopback bind logs a warning but is allowed.
+/// Flattened onto [`crate::listen::Config::tcp`]. Plaintext carries no peer
+/// identity and no encryption, so bind it to loopback or a private interface; a
+/// non-loopback plaintext bind logs a warning but is allowed. With
+/// [`tls`](Self::tls) the listener serves the listen TLS certificate instead.
 // The derived arg group is named after the struct, so it needs an explicit id to
 // stay unique across the flattened sections.
 #[derive(usage::Args, Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
@@ -27,7 +30,7 @@ const WIRE_VERSION: qmux::Version = qmux::Version::QMux01;
 #[serde(deny_unknown_fields, default)]
 #[non_exhaustive]
 pub struct Config {
-	/// Bind a plaintext qmux TCP listener on this address.
+	/// Bind a qmux TCP listener on this address, plaintext unless [`Self::tls`].
 	#[usage(
 		long = "listen-tcp-bind",
 		name = "listen-tcp-bind",
@@ -37,6 +40,21 @@ pub struct Config {
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub bind: Option<net::SocketAddr>,
 
+	/// Serve TLS on the TCP listener with the listen certificate, for `tls://`
+	/// dials. It asks for no client certificate, so a peer on it authenticates
+	/// with a token rather than mTLS.
+	#[usage(
+		long = "listen-tcp-tls",
+		name = "listen-tcp-tls",
+		env = "MOQ_LISTEN_TCP_TLS",
+		setting = "listen.tcp.tls",
+		default_missing = "true",
+		num_args = 0..=1,
+		require_equals = true,
+	)]
+	#[serde(default, skip_serializing_if = "Option::is_none")]
+	pub tls: Option<bool>,
+
 	/// The released `--server-tcp-bind` spelling and its env var, folded in by
 	/// [`Config::resolved`].
 	#[usage(flatten)]
@@ -76,7 +94,7 @@ impl Config {
 	}
 }
 
-/// Errors specific to the plain-TCP qmux transport.
+/// Errors specific to the qmux TCP transport.
 #[derive(Debug, thiserror::Error)]
 #[non_exhaustive]
 pub enum Error {
@@ -86,11 +104,11 @@ pub enum Error {
 	#[error(transparent)]
 	Io(#[from] std::io::Error),
 
-	/// The `tcp://` URL had no host.
+	/// The `tcp://` or `tls://` URL had no host.
 	#[error("missing hostname")]
 	MissingHostname,
 
-	/// The `tcp://` URL had no port. Unlike `https`, there is no default.
+	/// The `tcp://` or `tls://` URL had no port. Unlike `https`, there is no default.
 	#[error("missing port")]
 	MissingPort,
 
@@ -152,6 +170,44 @@ pub(crate) async fn connect(
 	connect_addrs(candidates, protocols, failover_delay).await
 }
 
+/// Dial a `tls://host:port` URL: qmux over TLS over TCP, negotiating one of
+/// `protocols` as the TLS ALPN and verifying the server with `tls`.
+///
+/// Resolves and races addresses exactly like [`connect`], through the TLS and
+/// qmux handshakes. The port is required.
+pub(crate) async fn connect_tls(
+	url: Url,
+	protocols: &[&str],
+	tls: &crate::tls::Connect,
+	failover_delay: std::time::Duration,
+	resolution_delay: std::time::Duration,
+) -> crate::Result<qmux::Session> {
+	let host = url.host().ok_or(Error::MissingHostname)?;
+	let port = url.port().ok_or(Error::MissingPort)?;
+	let name = tls.host_name.clone().unwrap_or_else(|| match &host {
+		url::Host::Domain(name) => name.to_string(),
+		url::Host::Ipv4(ip) => ip.to_string(),
+		url::Host::Ipv6(ip) => ip.to_string(),
+	});
+	let client = qmux::tls::Client::new(std::sync::Arc::new(tls.build()?))
+		.with_protocols(protocols.iter().map(|&alpn| (alpn, &[WIRE_VERSION][..])))
+		.require_protocol();
+
+	tracing::debug!(peer = %crate::connect::Endpoint(&url), "connecting via TLS over TCP");
+	let candidates = crate::resolve::Candidates::resolve(host, port, resolution_delay);
+	Ok(crate::failover::race(candidates, failover_delay, |addr| {
+		let client = client.clone();
+		let name = name.clone();
+		async move {
+			client
+				.connect(addr, &name)
+				.await
+				.map_err(|err| Error::Connect(crate::error::message(err)))
+		}
+	})
+	.await?)
+}
+
 /// Dial `candidates` in Happy Eyeballs order, performing the qmux handshake on
 /// each attempt; the first session to complete wins.
 async fn connect_addrs(
@@ -172,11 +228,13 @@ async fn connect_addrs(
 	.await
 }
 
-/// Listens for incoming plain-TCP qmux connections on a TCP port.
+/// Listens for incoming qmux connections on a TCP port, plaintext or TLS.
 pub struct Listener {
 	listener: tokio::net::TcpListener,
 	protocols: Vec<String>,
 	health: crate::accept::Health,
+	/// Present when the listener serves TLS; it negotiates the ALPN itself.
+	tls: Option<qmux::tls::Server>,
 }
 
 impl Listener {
@@ -187,9 +245,17 @@ impl Listener {
 			listener,
 			protocols: Vec::new(),
 			health: crate::accept::Health::new("tcp"),
+			tls: None,
 		})
 	}
 
+	/// Serve TLS with `config`, whose ALPN list already names each
+	/// `qmux-01.<protocol>` pair to accept.
+	pub(crate) fn with_tls(mut self, config: std::sync::Arc<rustls::ServerConfig>) -> Self {
+		self.tls = Some(qmux::tls::Server::new(config));
+		self
+	}
+
 	/// A live handle to this listener's accept-loop health, for an embedder that
 	/// publishes it (see [`crate::accept`]).
 	pub fn accept_health(&self) -> crate::accept::Health {
@@ -222,7 +288,8 @@ impl Listener {
 		Ok(self.listener.local_addr()?)
 	}
 
-	/// Accept the next connection, performing the qmux handshake over plain TCP.
+	/// Accept the next connection, performing the TLS (when configured) and qmux
+	/// handshakes.
 	///
 	/// A failed `accept(2)` is handled here rather than yielded: it is classified,
 	/// counted, logged, and paced by [`accept_health`](Self::accept_health), then
@@ -243,11 +310,13 @@ impl Listener {
 		let (stream, addr) = self.accept_socket().await;
 		tracing::debug!(%addr, "accepted TCP connection");
 		let config = qmux::tcp::Config::new(WIRE_VERSION).protocols(self.protocols.iter().map(String::as_str));
+		let tls = self.tls.clone();
 		async move {
-			let session = config
-				.accept(stream)
-				.await
-				.map_err(|err| Error::Accept(crate::error::message(err)))?;
+			let session = match tls {
+				Some(tls) => tls.accept(stream).await,
+				None => config.accept(stream).await,
+			}
+			.map_err(|err| Error::Accept(crate::error::message(err)))?;
 			Ok((session, addr))
 		}
 	}
@@ -276,6 +345,65 @@ mod tests {
 	use std::time::Duration;
 	use web_transport_trait::Session as _;
 
+	/// TLS needs a TCP listener to serve on; asking for it alone is refused.
+	#[test]
+	fn tls_without_a_bind_is_refused() {
+		let mut listen = crate::listen::Config::default();
+		listen.tcp.tls = Some(true);
+		assert!(matches!(
+			listen.init(Default::default()),
+			Err(crate::Error::NoBackend(_))
+		));
+	}
+
+	/// A `tls://` dial carries qmux and the request target over TLS, and a client
+	/// that does not trust the certificate is refused.
+	#[cfg(feature = "aws-lc-rs")]
+	#[tokio::test]
+	async fn tls_carries_qmux_and_refuses_an_untrusted_certificate() {
+		let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+		let mut listen = crate::listen::Config::default();
+		listen.tcp.bind = Some("127.0.0.1:0".parse().unwrap());
+		listen.tcp.tls = Some(true);
+		listen.tls.generate = vec!["localhost".into()];
+		// A client CA meant for QUIC peers does not make the TCP listener ask for one.
+		listen.tls.root = vec!["/nonexistent/client-ca.pem".into()];
+		let mut server = listen.init(Default::default()).unwrap().listen().await.unwrap();
+		let port = server.tcp_local_addr().unwrap().port();
+		let url: Url = format!("tls://localhost:{port}/room?jwt=credential").parse().unwrap();
+		let accept = tokio::spawn(async move {
+			let request = server.accept().await.unwrap();
+			assert_eq!(request.transport(), crate::server::Transport::Tcp);
+			assert_eq!(request.path(), "/room");
+			assert_eq!(request.query(), Some("jwt=credential"));
+			let session = request.ok().await.unwrap();
+			(server, session)
+		});
+
+		let dial = |insecure: bool| {
+			let mut config = crate::connect::Config {
+				once: Some(true),
+				..Default::default()
+			};
+			config.tls.insecure = Some(insecure);
+			config
+				.init(Default::default())
+				.unwrap()
+				.with_reconnect(false)
+				.connect(url.clone())
+		};
+		let untrusted = tokio::time::timeout(Duration::from_secs(5), dial(false).established()).await;
+		assert!(matches!(untrusted, Ok(Err(_))), "an untrusted certificate must be refused");
+		let _session = tokio::time::timeout(Duration::from_secs(5), dial(true).established())
+			.await
+			.expect("trusted dial timed out")
+			.expect("trusted dial failed");
+		let (_server, _accepted) = tokio::time::timeout(Duration::from_secs(5), accept)
+			.await
+			.expect("accept timed out")
+			.expect("accept task panicked");
+	}
+
 	/// End-to-end failover: the preferred candidate blackholes (TEST-NET-1 never
 	/// answers, or is unroutable outright in a sandbox), so the race must fall
 	/// through to the loopback listener within the stagger delay.

From f8f628db17b865f24003d9c3a86cdfe24826b3cc Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 4 Oct 2026 21:09:18 -0700
Subject: [PATCH 6/8] quest: finish upstream links

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 quest/m1/cluster-routing/README.md    |  6 +--
 quest/m1/cluster-routing/routes.md    |  2 +-
 quest/m1/cluster-routing/selection.md |  1 -
 quest/m1/cluster-routing/transit.md   | 64 ---------------------------
 4 files changed, 4 insertions(+), 69 deletions(-)
 delete mode 100644 quest/m1/cluster-routing/transit.md

diff --git a/quest/m1/cluster-routing/README.md b/quest/m1/cluster-routing/README.md
index a8c5e0becd..c485f78568 100644
--- a/quest/m1/cluster-routing/README.md
+++ b/quest/m1/cluster-routing/README.md
@@ -51,8 +51,9 @@ vector the 2026-09-30 cache-tiers audit kept:
   routes learned on another upstream link, and a route learned upstream keeps
   a down-only mark across other links so a mesh with two uplinks cannot leak
   CDN routes back into the CDN. An edge marks its core links upstream; a
-  Starlink drone marks its CDN link upstream. See
-  [Upstream links](/quest/m1/cluster-routing/transit.md).
+  Starlink drone marks its CDN link upstream. The per-link mark landed on
+  today's path vector (`doc/bin/relay/cluster.md`); the down-only mark lands
+  with [Routes and announces](/quest/m1/cluster-routing/routes.md).
 - **One node id space, no cluster ids.** A customer cluster is nodes behind
   upstream or trusted links. Loop detection by cluster id would drop a
   partitioned swarm's traffic to itself through the CDN (BGP's partitioned-AS
@@ -126,7 +127,6 @@ Once every child has landed:
 
 ## Required
 
-- [Upstream links](/quest/m1/cluster-routing/transit.md) - a link marked upstream never receives routes learned on another upstream link, which builds edge tiers and drone uplinks from one rule
 - [Selection](/quest/m1/cluster-routing/selection.md) - a broadcast under overlapping prefixes routes to one origin deterministically, and same-path origins are one source
 - [Multi-CDN endpoints](/quest/m1/cluster-routing/multi-cdn.md) - an endpoint holds sessions to several CDNs, uses its preferred one, and fails over to the next
 - [Simulate the split](/quest/m1/cluster-routing/sim.md) - moq.pro's simulator compares the route layer with path vector before the wire is written
diff --git a/quest/m1/cluster-routing/routes.md b/quest/m1/cluster-routing/routes.md
index 6b5d57fe86..27342e921a 100644
--- a/quest/m1/cluster-routing/routes.md
+++ b/quest/m1/cluster-routing/routes.md
@@ -67,7 +67,7 @@ Decided 2026-10-01 (see the [line's decisions](/quest/m1/cluster-routing/README.
   No per-announce seqno.
 - Down-only bit: set on a route learned on an upstream link, kept across
   other links, and a route carrying it is never sent on an upstream link
-  ([Upstream links](/quest/m1/cluster-routing/transit.md)).
+  (extending the `upstream` link mark in `doc/bin/relay/cluster.md`).
 - A plain client with one link advertises a ROUTE for itself and the
   ANNOUNCEs it publishes; its node id is scoped to its session (shared
   identity across sessions is [Route trust](/quest/m1/cluster-routing/route-trust.md)).
diff --git a/quest/m1/cluster-routing/selection.md b/quest/m1/cluster-routing/selection.md
index 42d5cc4497..aaf0f9088d 100644
--- a/quest/m1/cluster-routing/selection.md
+++ b/quest/m1/cluster-routing/selection.md
@@ -51,5 +51,4 @@ no timestamp rewind.
 
 ## Required
 
-- [Upstream links](/quest/m1/cluster-routing/transit.md) - the edges and cores this spreads over
 - [Wildcard](/quest/m0/wildcard/README.md) - the longest-prefix rule and pool spread this builds on
diff --git a/quest/m1/cluster-routing/transit.md b/quest/m1/cluster-routing/transit.md
deleted file mode 100644
index 46bc11f2f6..0000000000
--- a/quest/m1/cluster-routing/transit.md
+++ /dev/null
@@ -1,64 +0,0 @@
-# [M] Upstream links
-
-## Goal
-
-A cluster link can be marked upstream, and a relay never re-advertises a
-route learned on one upstream link to another upstream link. Everything else
-still transits. That one rule builds moq.pro's edge tier (an edge marks its
-region's cores upstream, so it is never transit between them) and a drone
-uplink (a Starlink drone marks its CDN link upstream: the mesh reaches the
-CDN and the CDN reaches the mesh, but the drone never carries CDN traffic
-between CDN relays). Edge, core, and one-node region are an operator's names
-for configurations, not relay modes. Clients still see every route a relay
-knows.
-
-## Plan
-
-Decided 2026-10-01, replacing the per-node `cluster.transit` setting (#4683)
-with a per-link attribute, because a drone must transit for its mesh but never
-for its CDN uplink, which no per-node switch can express (BGP's
-customer/provider export rule, RFC 9234, reduced to one bit):
-
-- `upstream` is an attribute of a peer entry, from config or the connect
-  API's peer list, off by default. It applies to the link whichever side
-  dialed: an edge that is dialed by a core still treats that core as
-  upstream, so accepted peer sessions need a way to be marked (the peer's
-  grant classification or identity is the likely source; pick the simplest
-  shape that covers the reverse-dial test).
-- An upstream link is sent only routes that did not arrive on another
-  upstream link. On today's path vector that is the routes that entered
-  locally or over non-upstream links.
-- The down-only mark that stops a two-uplink mesh from leaking CDN routes
-  back into the CDN needs a wire bit, so it lands with
-  [Routes and announces](/quest/m1/cluster-routing/routes.md), not here.
-- No roles and no topology check. Who dials whom is the peer list, and
-  whatever generates it applies the operator's rules (an edge dials its
-  region's cores, cores dial only lower-named cores of linked regions); the
-  generator is where that is tested.
-- Refusing clients on a hidden core is admission, not routing: the auth
-  server or embedder refuses a session that is not a cluster peer.
-- Relays from before this attribute treat every link as transit, so a cluster
-  migrates one region at a time with no legacy mode.
-- Spreading paths over equally ranked upstream links by the path-keyed
-  rendezvous hash is [Selection](/quest/m1/cluster-routing/selection.md)'s;
-  it is what makes every edge of a region send a given path to the same core.
-- Edge links run qmux over TLS on TCP, not WebSocket and not QUIC:
-  intra-region links are not congested. qmux over TCP exists today only in
-  plaintext (`tcp://`, moq-tokio's `tcp.rs`); add `tls://` beside it on
-  moq-tokio's `tls` helpers. (Scheme name decided 2026-09-30.) The closed
-  [#4731](https://github.com/moq-dev/moq/pull/4731) is prior art for the
-  `tls://` qmux piece in moq-tokio; its edge/core roles are not wanted.
-
-Test: a two-region cluster in `rs/moq-relay/tests`. One region has two edges
-whose links to two cores are upstream; the other is one relay that also
-serves clients. Subscribers on both edges cause one cross-region subscription
-per broadcast, a route from one core never reaches the other through an edge,
-a client on the far relay gets a broadcast published on an edge, a core that
-dials an edge still gets no routes through it, and losing a core moves only
-its paths. Add a three-node line (mesh node, uplink node, CDN relay) where the
-mesh node's broadcast reaches the CDN and a CDN broadcast reaches the mesh
-node.
-
-Public API: an `upstream` peer attribute and a TLS qmux URL scheme.
-`cluster.tier` is unrelated: it stays the billing and stats label.
-Wire: none.

From bd49ad7fda7a6c7348ac7bec95cdd22f453a99ce Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Sun, 4 Oct 2026 21:33:15 -0700
Subject: [PATCH 7/8] style: format

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-relay/src/cluster.rs          |  4 ++--
 rs/moq-relay/tests/upstream_links.rs | 15 ++++++++++++---
 rs/moq-tokio/src/tcp.rs              |  5 ++++-
 3 files changed, 18 insertions(+), 6 deletions(-)

diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs
index 184365a434..d01ea045c3 100644
--- a/rs/moq-relay/src/cluster.rs
+++ b/rs/moq-relay/src/cluster.rs
@@ -2461,8 +2461,8 @@ mod tests {
 	/// and flipping it is a different configuration, so the API redials.
 	#[test]
 	fn peer_object_upstream() {
-		let object: Peer = serde_json::from_str(r#"{"url": "https://core.example/", "upstream": true}"#)
-			.expect("parse upstream peer");
+		let object: Peer =
+			serde_json::from_str(r#"{"url": "https://core.example/", "upstream": true}"#).expect("parse upstream peer");
 		assert_eq!(object, Peer::new("https://core.example/").with_upstream(true));
 		assert_eq!(
 			serde_json::to_string(&object).unwrap(),
diff --git a/rs/moq-relay/tests/upstream_links.rs b/rs/moq-relay/tests/upstream_links.rs
index 03f3fa8074..32cf6f2fac 100644
--- a/rs/moq-relay/tests/upstream_links.rs
+++ b/rs/moq-relay/tests/upstream_links.rs
@@ -257,8 +257,14 @@ async fn tiered_regions_inner() {
 	let paths: Vec<String> = (0..8).map(|i| format!("live/{i}")).collect();
 	let paths: Vec<&str> = paths.iter().map(String::as_str).collect();
 	let _published: Vec<_> = paths.iter().map(|path| publish(&r.origin, path)).collect();
-	routes_when("both cores learn the far region", &c1.origin, |routes| has(routes, &paths)).await;
-	routes_when("both cores learn the far region", &c2.origin, |routes| has(routes, &paths)).await;
+	routes_when("both cores learn the far region", &c1.origin, |routes| {
+		has(routes, &paths)
+	})
+	.await;
+	routes_when("both cores learn the far region", &c2.origin, |routes| {
+		has(routes, &paths)
+	})
+	.await;
 
 	// A route is delivered before a later one: once an edge holds both cores'
 	// markers, it holds both cores' routes to the far region too.
@@ -285,7 +291,10 @@ async fn tiered_regions_inner() {
 	// Losing a core moves only its paths.
 	let on = |core_id: u64| -> Vec<&str> { paths.iter().copied().filter(|p| core(&at_e1, p) == core_id).collect() };
 	let (on_c1, on_c2) = (on(C1), on(C2));
-	assert!(!on_c1.is_empty() && !on_c2.is_empty(), "paths all on one core: {at_e1:?}");
+	assert!(
+		!on_c1.is_empty() && !on_c2.is_empty(),
+		"paths all on one core: {at_e1:?}"
+	);
 	drop(c1);
 	let after = routes_when("edge 1 drops the lost core", &e1.origin, |routes| {
 		has(routes, &paths) && routes.values().all(|hops| !hops.contains(&C1))
diff --git a/rs/moq-tokio/src/tcp.rs b/rs/moq-tokio/src/tcp.rs
index 09d4db8653..83e724267b 100644
--- a/rs/moq-tokio/src/tcp.rs
+++ b/rs/moq-tokio/src/tcp.rs
@@ -393,7 +393,10 @@ mod tests {
 				.connect(url.clone())
 		};
 		let untrusted = tokio::time::timeout(Duration::from_secs(5), dial(false).established()).await;
-		assert!(matches!(untrusted, Ok(Err(_))), "an untrusted certificate must be refused");
+		assert!(
+			matches!(untrusted, Ok(Err(_))),
+			"an untrusted certificate must be refused"
+		);
 		let _session = tokio::time::timeout(Duration::from_secs(5), dial(true).established())
 			.await
 			.expect("trusted dial timed out")

From 620ba4c78f51437fe9b3d440f7f2ebb420f95f91 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Mon, 5 Oct 2026 23:48:06 -0700
Subject: [PATCH 8/8] fix(tokio): the tls:// TCP listener ignores pinned peers
 too

Clearing only `root` left a QUIC pinned-peer allowlist on the TCP
listener's TLS config, so a token-only tls:// client was refused with
CertificateRequired before SETUP. Cover both policies in the test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-tokio/src/server.rs |  5 +++--
 rs/moq-tokio/src/tcp.rs    | 25 ++++++++++++++++++-------
 2 files changed, 21 insertions(+), 9 deletions(-)

diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs
index 871f4ea7e6..0e4ab43fa2 100644
--- a/rs/moq-tokio/src/server.rs
+++ b/rs/moq-tokio/src/server.rs
@@ -1014,11 +1014,12 @@ impl TcpTls {
 	/// `qmux-01.<alpn>` TLS ALPN.
 	///
 	/// Asks for no client certificate: qmux's TLS accept keeps no peer identity
-	/// for the auth server to read, so a `tls.root` meant for QUIC peers does not
-	/// apply, and a peer on this listener authenticates with a token.
+	/// for the auth server to read, so neither a `tls.root` nor pinned `peers`
+	/// meant for QUIC apply, and a peer on this listener authenticates with a token.
 	fn new(config: &crate::listen::Config, versions: &moq_net::Versions) -> crate::Result<Self> {
 		let mut listen = config.tls.clone();
 		listen.root.clear();
+		listen.peers = None;
 		let alpn = stream_versions(versions)
 			.alpns()
 			.iter()
diff --git a/rs/moq-tokio/src/tcp.rs b/rs/moq-tokio/src/tcp.rs
index 83e724267b..853dc883b8 100644
--- a/rs/moq-tokio/src/tcp.rs
+++ b/rs/moq-tokio/src/tcp.rs
@@ -357,17 +357,28 @@ mod tests {
 	}
 
 	/// A `tls://` dial carries qmux and the request target over TLS, and a client
-	/// that does not trust the certificate is refused.
+	/// that does not trust the certificate is refused. Client-certificate policy
+	/// meant for QUIC (CA roots or pinned peers) never makes the TCP listener ask
+	/// for one, so a token-only client still gets through.
 	#[cfg(feature = "aws-lc-rs")]
 	#[tokio::test]
 	async fn tls_carries_qmux_and_refuses_an_untrusted_certificate() {
 		let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
-		let mut listen = crate::listen::Config::default();
-		listen.tcp.bind = Some("127.0.0.1:0".parse().unwrap());
-		listen.tcp.tls = Some(true);
-		listen.tls.generate = vec!["localhost".into()];
-		// A client CA meant for QUIC peers does not make the TCP listener ask for one.
-		listen.tls.root = vec!["/nonexistent/client-ca.pem".into()];
+		for pinned in [false, true] {
+			let mut listen = crate::listen::Config::default();
+			listen.tcp.bind = Some("127.0.0.1:0".parse().unwrap());
+			listen.tcp.tls = Some(true);
+			listen.tls.generate = vec!["localhost".into()];
+			match pinned {
+				true => listen.tls.peers = Some(crate::tls::Peers::new()),
+				false => listen.tls.root = vec!["/nonexistent/client-ca.pem".into()],
+			}
+			tls_round_trip(listen).await;
+		}
+	}
+
+	#[cfg(feature = "aws-lc-rs")]
+	async fn tls_round_trip(listen: crate::listen::Config) {
 		let mut server = listen.init(Default::default()).unwrap().listen().await.unwrap();
 		let port = server.tcp_local_addr().unwrap().port();
 		let url: Url = format!("tls://localhost:{port}/room?jwt=credential").parse().unwrap();