fix(net): give an anonymous dial its own hop (backport #5025)

rs/moq-net/src/client.rs

@@ -108,41 +108,47 @@ impl Client {
 		self
 	}
 
-	/// Assign an origin (hop) id to the peer, used whenever the peer doesn't declare
-	/// one itself.
+	/// Assign the identity this peer's routes are attributed to, overriding the fresh
+	/// per-dial default.
 	///
-	/// Some relays never declare their identity: moq-lite peers without the hops
-	/// extension, and moq-transport peers that don't negotiate the MoQ Cluster
-	/// extension (or predate it, on `moqt-16` and earlier).
-	/// Broadcasts received from such a peer are normally attributed to the reserved
-	/// Hop ID 0 ("unknown"), which identifies nothing: it never proves continuity,
-	/// so their advertisements neither splice nor survive a restart in place. This
-	/// knob pins a real identity instead, exactly as if the peer had declared it:
+	/// A dialed session whose peer declares no hop gets a random one for that
+	/// connection, the same way an accepted session does. That keeps a route
+	/// learned from the peer off the session that learned it, and keeps a
+	/// SUBSCRIBE that arrives on it from being routed back to it. An identity the
+	/// peer declares on the wire still wins.
 	///
-	/// - broadcasts received from the peer carry `origin` in their hop chains, so
-	///   every session dialing the same relay (with the same id) resolves to one
-	///   route and loop checks can recognize it;
-	/// - broadcasts whose hop chain already contains `origin` are neither announced
-	///   nor served back to the peer, preventing an echo through a relay that does
-	///   no loop detection of its own.
-	///
-	/// An identity the peer does declare wins over this one.
+	/// Pass an id only for a peer whose identity the caller has actually
+	/// established. Two dials given the same hop are treated as one endpoint:
+	/// routes learned from either are kept off both, and content arriving on
+	/// either is interchangeable with the other's. That is the point when they
+	/// really are one peer reconnecting or running redundant links, and a bug
+	/// otherwise. Derive it from the authenticated identity, never from something
+	/// coarser like the remote address.
 	pub fn with_peer_hop(mut self, hop: crate::Hop) -> Self {
 		self.peer_hop = Some(hop);
 		self
 	}
 
+	/// The hop this connection attributes the peer to when the peer declares none.
+	///
+	/// A caller-supplied id is stable across dials of this client. Otherwise each
+	/// connection gets a fresh one, matching the per-session default an accepted
+	/// session already has.
+	fn assigned_hop(&self) -> crate::Hop {
+		self.peer_hop.unwrap_or_else(crate::Hop::random)
+	}
+
 	/// The origin pair a session attaches, tagged and filtered.
 	///
 	/// Reads through the publish (egress) consumer and writes through the
 	/// subscribe (ingress) producer are attributed by the model through the
 	/// stats context; one shared context, so presence and viewer counts are
-	/// never double-attributed across the two halves. An assigned peer identity
-	/// means subscriptions from the peer resolve to a source whose hop chain
-	/// excludes it, the same split-horizon rule applied when a peer declares
-	/// its own id; announce filtering is per-protocol and handled inside each
-	/// publisher.
-	fn origins(&self) -> (Option<origin::Consumer>, Option<origin::Producer>) {
+	/// never double-attributed across the two halves. `peer_hop` is the identity
+	/// this connection attributes the peer to when the peer declares none, so
+	/// subscriptions from the peer resolve to a source whose route excludes it.
+	/// A declared identity still wins inside each publisher; announce filtering
+	/// is per-protocol and handled there too.
+	fn origins(&self, peer_hop: crate::Hop) -> (Option<origin::Consumer>, Option<origin::Producer>) {
 		if self.publish.is_none() && self.subscribe.is_none() {
 			tracing::warn!("not publishing or consuming anything");
 		}
@@ -151,7 +157,7 @@ impl Client {
 			.subscribe
 			.clone()
 			.map(|origin| origin.with_stats(self.stats.clone()));
-		let publish = publish.map(|origin| origin.excluding(self.peer_hop.unwrap_or(crate::Hop::UNKNOWN)));
+		let publish = publish.map(|origin| origin.excluding(peer_hop));
 		(publish, subscribe)
 	}
 
@@ -162,11 +168,12 @@ impl Client {
 		runtime: Clock,
 		session: S,
 		version: lite::Version,
+		peer_hop: crate::Hop,
 	) -> Result<(Session, crate::Driver<S>), Error>
 	where
 		S: crate::transport::poll::Session,
 	{
-		let (publish, subscribe) = self.origins();
+		let (publish, subscribe) = self.origins(peer_hop);
 
 		// Advertise our capabilities (we report what the transport measures; we
 		// don't pad) plus the request path on URI-less transports, and the
@@ -192,7 +199,7 @@ impl Client {
 			setup_stream: None,
 			publish,
 			subscribe,
-			peer_hop: self.peer_hop,
+			peer_hop: Some(peer_hop),
 			version,
 			our_setup,
 			peer_setup: None,
@@ -231,7 +238,7 @@ impl Client {
 			_ => return Err(Error::Version),
 		};
 		self.versions.select(Version::Lite(version)).ok_or(Error::Version)?;
-		self.start_lite(runtime, session, version)
+		self.start_lite(runtime, session, version, self.assigned_hop())
 	}
 
 	/// Perform the MoQ handshake, returning the [`Session`] and its [`Driver`](crate::Driver).
@@ -242,7 +249,8 @@ impl Client {
 		S: crate::transport::poll::Boxable,
 	{
 		let runtime = Clock::new(now);
-		let (publish, subscribe) = self.origins();
+		let peer_hop = self.assigned_hop();
+		let (publish, subscribe) = self.origins(peer_hop);
 
 		// If ALPN was used to negotiate the version, use the appropriate encoding.
 		// Default to IETF 14 if no ALPN was used and we'll negotiate the version later.
@@ -269,7 +277,7 @@ impl Client {
 					client: true,
 					publish: publish.clone(),
 					subscribe: subscribe.clone(),
-					peer_hop: self.peer_hop,
+					peer_hop: Some(peer_hop),
 					cost: self.cost,
 					version: draft,
 					path: self.setup_path.clone(),
@@ -317,19 +325,19 @@ impl Client {
 					_ => lite::Version::Lite05,
 				};
 				self.versions.select(Version::Lite(version)).ok_or(Error::Version)?;
-				return self.start_lite(runtime, session, version);
+				return self.start_lite(runtime, session, version, peer_hop);
 			}
 			Some(ALPN_LITE_04) => {
 				self.versions
 					.select(Version::Lite(lite::Version::Lite04))
 					.ok_or(Error::Version)?;
-				return self.start_lite(runtime, session, lite::Version::Lite04);
+				return self.start_lite(runtime, session, lite::Version::Lite04, peer_hop);
 			}
 			Some(ALPN_LITE_03) => {
 				self.versions
 					.select(Version::Lite(lite::Version::Lite03))
 					.ok_or(Error::Version)?;
-				return self.start_lite(runtime, session, lite::Version::Lite03);
+				return self.start_lite(runtime, session, lite::Version::Lite03, peer_hop);
 			}
 			Some(ALPN_LITE) | None => {
 				let supported = self.versions.filter(&NEGOTIATED.into()).ok_or(Error::Version)?;
@@ -381,7 +389,7 @@ impl Client {
 					setup_stream: Some(stream),
 					publish: publish.clone(),
 					subscribe: subscribe.clone(),
-					peer_hop: self.peer_hop,
+					peer_hop: Some(peer_hop),
 					version: v,
 					// This path only handles versions negotiated via the bidi SETUP exchange
 					// (pre-lite-05), which have no Setup Stream.
@@ -418,7 +426,7 @@ impl Client {
 					client: true,
 					publish: publish.clone(),
 					subscribe: subscribe.clone(),
-					peer_hop: self.peer_hop,
+					peer_hop: Some(peer_hop),
 					cost: self.cost,
 					version: v,
 					path: None,

rs/moq-net/src/ietf/publisher.rs

@@ -313,8 +313,8 @@ pub(super) struct Publisher<S: crate::transport::poll::Session> {
 	// origin we consume so it matches the local relay identity across every session,
 	// which is what makes cross-session loop detection work.
 	self_origin: crate::Hop,
-	// The identity assigned to the peer by the caller (`Client::with_peer_hop`, or
-	// the per-session default a server hands every request), used when the peer declares
+	// The identity assigned to the peer (a fresh per-session id, or one the caller
+	// pinned with `with_peer_hop`), used when the peer declares
 	// none itself. A peer that negotiates the MoQ Cluster extension declares its own,
 	// which wins unless it withheld it as the reserved 0.
 	peer_hop: Option<crate::Hop>,
@@ -420,8 +420,8 @@ where
 	/// The Hop ID whose paths must not be advertised (or served) back to this peer.
 	///
 	/// A peer that declared an identity supplies its own; otherwise fall back to the one
-	/// we assigned it (`Client::with_peer_hop` when dialing, `Request::with_peer_hop`
-	/// or a fresh per-session id when accepting), since moq-transport carries no identity
+	/// we assigned it (a fresh per-session id, or one pinned with `Client::with_peer_hop`
+	/// or `Request::with_peer_hop`), since moq-transport carries no identity
 	/// of its own. A peer that declared the reserved 0 declared no identity, so it takes
 	/// the fallback like any other anonymous peer.
 	fn exclude(&self, peer: &cluster::Peer) -> crate::Hop {

rs/moq-net/src/ietf/subscriber.rs

@@ -467,12 +467,12 @@ pub(super) struct Subscriber<S: crate::transport::poll::Session> {
 	// The origin naming this link for split-horizon (`Route.via`) when the peer
 	// declares none of its own (see `session_route`). Base moq-transport carries no
 	// hop ids, so a peer only has an identity if it negotiated the MoQ Cluster
-	// extension or the caller assigned it one (`Client::with_peer_hop`).
+	// extension; otherwise this is the one the session assigned it (a fresh id per
+	// dialed or accepted session, unless the caller pinned one with `with_peer_hop`),
+	// or `Hop::UNKNOWN` (0) when none was assigned.
 	//
-	// Otherwise this is `Hop::UNKNOWN` (0), the reserved "no identity" value.
 	// The assigned id stays local: it is never written into a hop chain, so a peer
-	// that withheld an identity is not named on the wire. A server answers it per
-	// accepted session; a client only when it knows the peer.
+	// that withheld an identity is not named on the wire.
 	session_origin: crate::Hop,
 	// A random Hop ID of this connection's own, written as the first hop of any path
 	// that arrives naming no publisher, so a publisher that reconnects reads downstream

rs/moq-net/src/lite/publisher.rs

@@ -51,8 +51,8 @@ struct Shared<S: crate::transport::poll::Session> {
 	// peer from a source whose chain excludes them, keeping the data plane on
 	// the same split-horizon rule as the announces we send them.
 	peer_setup: super::PeerSetup,
-	// The identity assigned to the peer by the caller (`Client::with_peer_hop`, or
-	// the per-session default a server hands every request), standing in wherever the
+	// The identity assigned to the peer (a fresh per-session id, or one the caller
+	// pinned with `with_peer_hop`), standing in wherever the
 	// peer declines to declare one. Backs both the announce filter and the serving
 	// origin, so a peer that names itself nowhere on the wire is still split-horizoned.
 	peer_hop: Option<Hop>,
@@ -528,8 +528,7 @@ impl<S: crate::transport::poll::Session> AnnounceServe<S> {
 					// The identity whose routes we filter out. Lite-04/05 carry it per
 					// announce stream; lite-06+ reads the session-wide SETUP Hop
 					// parameter, the same identity the subscribe path excludes. A peer that
-					// declares nothing falls back to the identity the caller assigned it
-					// (`with_peer_hop`), if any.
+					// declares nothing falls back to the identity the session assigned it.
 					let assigned = self.shared.peer_hop.map(|origin| origin.id()).unwrap_or(0);
 					if self.shared.version.has_exclude_hop() {
 						let exclude_hop = match interest.exclude_hop {

rs/moq-net/src/lite/subscriber.rs

@@ -61,13 +61,11 @@ pub(super) struct Subscriber<S: crate::transport::poll::Session> {
 	// carry real hop ids on the wire (Lite01/02/03), and for a peer that reports
 	// 0 in AnnounceOk. Lite03 placeholders stay 0 and count as anonymous.
 	//
-	// This is the peer's assigned identity (`peer_hop`) when the caller gave
-	// it one. Otherwise it is `Hop::UNKNOWN` (0), the reserved "no identity" value.
-	//
-	// Assigning one is the caller's call, not this layer's: a server gives every
-	// accepted session a fresh id so its routes are at least distinguishable from
-	// another session's, while a client only assigns one it knows out of band. The
-	// assigned id stays local and is never written into a hop chain.
+	// This is the peer's assigned identity (`peer_hop`): a fresh id per dialed or
+	// accepted session, so its routes are distinguishable from another session's,
+	// unless the caller pinned a stable one with `with_peer_hop`. Without one it is
+	// `Hop::UNKNOWN` (0), the reserved "no identity" value. The assigned id stays
+	// local and is never written into a hop chain.
 	session_origin: crate::Hop,
 	// A random Hop ID of this connection's own, written as the first hop of any chain
 	// that names no publisher (Lite01/02, a Lite03 placeholder, or a peer reporting 0),