perf(net): lite-07 announce streams cost each claim change, not every claim held

quest/m2/cluster-routing/routes/README.md

@@ -542,7 +542,7 @@ benchmarks.
 - [Route layer acceptance](/quest/m2/cluster-routing/routes/acceptance.md) - forwarding fixtures ported and the model deleted, incident fixtures, end-to-end checks, and benchmarks
 - [Firefox tabs prove their route key](/quest/m2/cluster-routing/routes/browser-proof.md) - a current Firefox tab opens a lite-07 session with a route-key proof the relay verifies, once neqo's exporter follows the draft
 - [Playwright bundles Chromium's spec exporter](/quest/m2/cluster-routing/routes/playwright-pin.md) - the toolchain's Playwright bundles Chromium 158+, so the route key lane proves the Chromium tab
-- [Showing an identity's claims costs the claims that changed](/quest/m2/cluster-routing/routes/show-cost.md) - lite-07 session setup and teardown grow linearly with broadcasts, not quadratically per identity
+- [The route node behaves the same whoever creates it](/quest/m2/cluster-routing/routes/node-creation-order.md) - creating an origin's lite-07 route node from SETUP instead of the route layer changes nothing observable
 
 ## Related
 

rs/moq-net/benches/session/mod.rs

@@ -43,6 +43,10 @@
 //! `session_join_*` times a new viewer connecting, resolving a broadcast, and
 //! receiving its latest group, swept over what the relays already announce.
 //!
+//! `session_announce` times one broadcast announced and withdrawn beside many a publisher
+//! already announces, until every viewer session saw both, swept over those broadcasts and
+//! the viewers: a change's cost grows with the viewers it reaches, never with the claims held.
+//!
 //! `session_dash_*` models the moq.pro dash aggregator: one session following
 //! every node's per-project stats broadcast, each announced by a peer relay, at
 //! one small frame per track per tick. Throughput counts tracks, so a flat
@@ -601,7 +605,8 @@ struct Cluster<B: Backend> {
 	/// One datagram track per broadcast, indexed like [`path`], for datagram shapes only.
 	datagrams: Vec<datagram::Producer>,
 	_broadcasts: Vec<origin::Route>,
-	_origins: Vec<origin::Producer>,
+	/// Each publisher's origin, indexed like `publisher`.
+	origins: Vec<origin::Producer>,
 	_pairs: Vec<Pair>,
 	next_hop: u64,
 	/// The relay and publisher caches; each viewer owns its own.
@@ -621,7 +626,7 @@ impl<B: Backend> Cluster<B> {
 			tracks: Vec::new(),
 			datagrams: Vec::new(),
 			_broadcasts: Vec::new(),
-			_origins: Vec::new(),
+			origins: Vec::new(),
 			_pairs: Vec::new(),
 			next_hop: 0,
 			pools: Vec::new(),
@@ -689,7 +694,7 @@ impl<B: Backend> Cluster<B> {
 			let target = &this.relays[publisher % shape.relays];
 			options.server_subscribe = Some(B::origin(target).clone());
 			this._pairs.push(B::connect(options, target, false).await);
-			this._origins.push(origin);
+			this.origins.push(origin);
 		}
 
 		this
@@ -1477,6 +1482,69 @@ fn join<B: Backend>(c: &mut Criterion, name: &str, shapes: impl IntoIterator<Ite
 	group.finish();
 }
 
+/// `(broadcasts, viewers)` shapes for [`announce`].
+const ANNOUNCE: [(usize, usize); 4] = [(16, 1), (4096, 1), (16, 64), (4096, 64)];
+
+/// One more broadcast announced and withdrawn beside `broadcasts` a publisher already
+/// announces through one relay, timed until each of `viewers` sessions saw both.
+fn announce<B: Backend>(c: &mut Criterion) {
+	let mut group = c.benchmark_group(format!("{}session_announce", B::PREFIX));
+	for version in VERSIONS {
+		for (broadcasts, viewers) in ANNOUNCE {
+			let rt = runtime();
+			let mut room = None;
+			let id = format!("{version}/broadcasts={broadcasts}/viewers={viewers}");
+			group.bench_function(BenchmarkId::from_parameter(id), |b| {
+				let (cluster, routes, next) = room.get_or_insert_with(|| {
+					rt.block_on(async {
+						let shape = Shape {
+							publishers: 1,
+							broadcasts,
+							viewers,
+							..Shape::BASE
+						};
+						let mut cluster = Cluster::<B>::new(version, shape, false).await;
+						let mut routes = Vec::new();
+						for _ in 0..viewers {
+							let viewer = cluster.join(0, false).await;
+							let mut announced = viewer.origin.consume().routes();
+							for _ in 0..broadcasts {
+								let event = announced.next().await.expect("announced");
+								assert_eq!(event.kind, origin::RouteKind::Start);
+							}
+							routes.push((viewer, announced));
+						}
+						(cluster, routes, broadcasts)
+					})
+				});
+				b.iter_custom(|iters| {
+					rt.block_on(async {
+						let start = Instant::now();
+						for _ in 0..iters {
+							// A fresh path each time: a broadcast's identity never comes back.
+							let route = cluster.origins[0]
+								.route(path(*next), origin::RouteOptions::default())
+								.unwrap();
+							*next += 1;
+							for (_, announced) in routes.iter_mut() {
+								let event = announced.next().await.expect("announced");
+								assert_eq!(event.kind, origin::RouteKind::Start);
+							}
+							drop(route);
+							for (_, announced) in routes.iter_mut() {
+								let event = announced.next().await.expect("withdrawn");
+								assert_eq!(event.kind, origin::RouteKind::End);
+							}
+						}
+						start.elapsed()
+					})
+				})
+			});
+		}
+	}
+	group.finish();
+}
+
 /// The moq.pro dash aggregator's load: one session on relay 0 following every
 /// node's per-project stats broadcast. Every other relay is a node publishing
 /// `.stats/<project>/node/<node>` into its own origin, as `moq-relay`'s stats
@@ -2101,6 +2169,8 @@ pub(crate) fn session<B: Backend>(c: &mut Criterion) {
 
 	join::<B>(c, "relays", [1, 2, 8].map(|relays| Shape { relays, ..base }));
 
+	announce::<B>(c);
+
 	churn::<B>(c, "sessions", [1, 16, 128].map(|sessions| Churn { sessions, held: 16 }));
 	churn::<B>(
 		c,

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

@@ -39,6 +39,8 @@ struct Sent {
 	reach: BTreeMap<usize, route::Reach>,
 	/// Every claim the node sent for it, named as the handle names it, before filtering.
 	claims: BTreeMap<Path, node::Claim>,
+	/// How many of `claims` the stream's link never carries (see [`sends`]).
+	withheld: usize,
 	/// The node's floor for those claims.
 	floor: Option<route::Rev>,
 	/// Whether the peer was sent a floor for it yet.
@@ -49,6 +51,17 @@ impl Sent {
 	fn reach(&self) -> Option<route::Reach> {
 		self.reach.values().next().copied()
 	}
+
+	/// The node's claim at `path` is now `claim`, which `link` may withhold.
+	fn set(&mut self, link: &route::Link, path: Path, claim: Option<node::Claim>) {
+		let withheld = claim.as_ref().is_some_and(|claim| !sends(link, claim));
+		let old = match claim {
+			Some(claim) => self.claims.insert(path, claim),
+			None => self.claims.remove(&path),
+		};
+		self.withheld += usize::from(withheld);
+		self.withheld -= usize::from(old.is_some_and(|old| !sends(link, &old)));
+	}
 }
 
 /// A node stream the answer reads.
@@ -339,6 +352,7 @@ impl Answer {
 					shown: 0,
 					reach: BTreeMap::from([(source, reach)]),
 					claims: BTreeMap::new(),
+					withheld: 0,
 					floor: None,
 					floored: false,
 				};
@@ -377,7 +391,7 @@ impl Answer {
 				self.echo(id, source);
 				let sent = self.routes.get_mut(&id).expect("matched above");
 				for path in &paths {
-					sent.claims.remove(path);
+					sent.set(&self.link, path.clone(), None);
 				}
 				for path in paths {
 					self.sync(stream, id, path)?;
@@ -502,10 +516,7 @@ impl Answer {
 				None if self.origin.shadows(&path) => continue,
 				None => path,
 			};
-			match claim {
-				Some(claim) => sent.claims.insert(path.clone(), claim),
-				None => sent.claims.remove(&path),
-			};
+			sent.set(&self.link, path.clone(), claim);
 			paths.push(path);
 		}
 		sent.floor = floor;
@@ -521,10 +532,8 @@ impl Answer {
 		let certified = self.certified();
 		// Like the grant, the upstream horizon and the proxy preference leave claims out
 		// that a floor would certify absent.
-		let hides = (self.routes.get(&id).into_iter())
-			.flat_map(|sent| sent.claims.values())
-			.any(|claim| !sends(&self.link, claim));
 		let sent = self.routes.get_mut(&id).expect("only a held route has a batch");
+		let hides = sent.withheld > 0;
 		let Some(wire) = sent.wire else {
 			return Ok(());
 		};

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

@@ -6,7 +6,7 @@
 //! the origin, minting a source per request that names the claim's origin node.
 
 use std::{
-	collections::{BTreeMap, BTreeSet, HashMap},
+	collections::{BTreeMap, HashMap},
 	task::Poll,
 };
 
@@ -22,8 +22,8 @@ use super::{AnnouncedRoute, Subscriber};
 struct WireRoute {
 	node: route::Id,
 	reach: route::Reach,
-	/// The Announce IDs of its live claims.
-	claims: BTreeSet<u64>,
+	/// The Announce ID of each of its live claims, by path.
+	claims: BTreeMap<Path, u64>,
 	/// How its claims differ from `committed`, by path, so what the stream keeps
 	/// between floors is bounded by its claims rather than the messages it read.
 	pending: BTreeMap<Path, Option<node::Claim>>,
@@ -175,7 +175,7 @@ impl Routed {
 					WireRoute {
 						node,
 						reach,
-						claims: BTreeSet::new(),
+						claims: BTreeMap::new(),
 						pending: BTreeMap::new(),
 						committed: None,
 					},
@@ -186,7 +186,7 @@ impl Routed {
 			Msg::RouteHold { id } => self.reach(id, route::Reach::Held)?,
 			Msg::RouteEnd { id } => {
 				let route = self.routes.remove(&id).ok_or(Error::ProtocolViolation)?;
-				for claim in route.claims {
+				for claim in route.claims.into_values() {
 					decoder.end(claim)?;
 					self.claims.remove(&claim);
 					self.held.pop();
@@ -271,17 +271,12 @@ impl Routed {
 
 	/// A claim the peer started under `route`, assigned Announce ID `id`.
 	fn start(&mut self, id: u64, route: u64, path: Path, claim: node::Claim) -> Result<(), Error> {
-		let claims = &self.claims;
 		let wire = self.routes.get_mut(&route).ok_or(Error::ProtocolViolation)?;
-		if wire
-			.claims
-			.iter()
-			.any(|other| claims.get(other).is_some_and(|other| other.path == path))
-		{
+		if wire.claims.contains_key(&path) {
 			return Err(Error::ProtocolViolation);
 		}
 		self.held.push(self.slots.acquire()?);
-		wire.claims.insert(id);
+		wire.claims.insert(path.clone(), id);
 		wire.change(path.clone(), Some(claim.clone()));
 		self.claims.insert(id, WireClaim { route, path, claim });
 		Ok(())
@@ -292,7 +287,7 @@ impl Routed {
 		let claim = self.claims.remove(&id).ok_or(Error::ProtocolViolation)?;
 		self.held.pop();
 		let route = self.routes.get_mut(&claim.route).ok_or(Error::ProtocolViolation)?;
-		route.claims.remove(&id);
+		route.claims.remove(&claim.path);
 		route.change(claim.path, None);
 		Ok(())
 	}