bench(moq-net): model the dash aggregator's stats load in the session bench

quest/m1/bench-relay.md

@@ -18,3 +18,9 @@ than the bench is worth, say so and stop.
 Reuse the scenario, the publisher and subscriber sweeps, and the delivery
 accounting from `rs/moq-net/benches/session.rs` so the two results line up, and
 the difference is the relay layer.
+
+Run the dash shape (`session_dash_*`) first. In the model bench, the production
+point (34 peer nodes x 5 projects x 24 tracks) costs about 100 ms of one core
+per 1 s stats tick, for the whole mesh. In production, moving that one session
+off a relay saved 50-90% of a core on that relay alone (2026-09-25). The gap is
+the relay layer or the transport, not the model.

rs/moq-net/benches/session.rs

@@ -15,6 +15,11 @@
 //! `session_join_*` times a new viewer connecting, resolving a broadcast, and
 //! receiving its latest group, swept over what the relays already announce.
 //!
+//! `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
+//! elements/s across a sweep means the cost per track stays constant.
+//!
 //! Everything runs on one current-thread runtime, so the only work measured is
 //! the protocol and model code, never scheduling across threads.
 //!
@@ -41,6 +46,12 @@ const FRAMES: usize = 4;
 
 const TRACK: &str = "video";
 
+/// Lite 06 is what the dash session and cluster peers negotiate today.
+const DASH_VERSIONS: [&str; 2] = ["moq-lite-06", "moq-transport-22"];
+
+/// A stats frame: a small JSON snapshot.
+const DASH_FRAME: usize = 128;
+
 /// Each endpoint gets its own bounded pool, as a relay configures one per
 /// process: without a byte target, every hop keeps each group for the whole
 /// expiry window and memory grows with the run length instead of the shape.
@@ -151,9 +162,9 @@ fn runtime() -> tokio::runtime::Runtime {
 		.unwrap()
 }
 
-fn write_group(track: &track::Producer, payload: &Bytes) {
+fn write_group(track: &track::Producer, payload: &Bytes, frames: usize) {
 	let mut group = track.append_group().unwrap();
-	for _ in 0..FRAMES {
+	for _ in 0..frames {
 		group.write_frame(Timestamp::ZERO, payload.clone()).unwrap();
 	}
 	group.finish().unwrap();
@@ -248,32 +259,38 @@ impl Cluster {
 		producer
 	}
 
-	/// Connect a viewer to `relay` and subscribe it to each of `broadcasts`.
-	async fn join(&mut self, relay: usize, broadcasts: impl Iterator<Item = usize>) -> Viewer {
+	/// Connect a viewer session to `relay`.
+	async fn join(&mut self, relay: usize) -> Viewer {
 		let origin = self.origin(CLIENT_CACHE);
 		let mut options = MockConnectOptions::new(self.version);
-		options.server_publish = Some(self.relays[relay].consume());
+		// Hidden included, as the dash's `.stats/` scope sees its subtree.
+		options.server_publish = Some(self.relays[relay].consume().with_hidden(true));
 		options.client_subscribe = Some(origin.clone());
 		let pair = connect_mock(options).await;
 
-		let mut subscribers = Vec::new();
-		for broadcast in broadcasts {
-			let broadcast = origin.consume().routed_broadcast(path(broadcast)).await.unwrap();
-			subscribers.push(broadcast.track(TRACK).unwrap().subscribe(None).await.unwrap());
-		}
-
 		Viewer {
-			subscribers,
+			subscribers: Vec::new(),
 			_pair: pair,
-			_origin: origin,
+			origin,
 		}
 	}
 }
 
 struct Viewer {
 	subscribers: Vec<track::Subscriber>,
 	_pair: MockPair,
-	_origin: origin::Producer,
+	origin: origin::Producer,
+}
+
+impl Viewer {
+	/// Resolve `path` and subscribe to each of its `tracks`.
+	async fn watch(&mut self, path: &str, tracks: &[impl AsRef<str>]) {
+		let broadcast = self.origin.consume().routed_broadcast(path).await.unwrap();
+		for track in tracks {
+			self.subscribers
+				.push(broadcast.track(track.as_ref()).unwrap().subscribe(None).await.unwrap());
+		}
+	}
 }
 
 /// A cluster with its viewers attached.
@@ -313,8 +330,11 @@ impl Room {
 			}
 			// Offset by one so a viewer lands on a different relay than the
 			// publisher it watches whenever there is more than one.
-			let relay = (viewer + 1) % shape.relays;
-			viewers.push(cluster.join(relay, broadcasts.into_iter()).await);
+			let mut viewer = cluster.join((viewer + 1) % shape.relays).await;
+			for broadcast in broadcasts {
+				viewer.watch(&path(broadcast), &[TRACK]).await;
+			}
+			viewers.push(viewer);
 		}
 
 		let mut room = Self {
@@ -346,7 +366,7 @@ impl Room {
 			return self.paced_round().await;
 		}
 		for &broadcast in &self.watched {
-			write_group(&self.cluster.tracks[broadcast], &self.payload);
+			write_group(&self.cluster.tracks[broadcast], &self.payload, FRAMES);
 		}
 		let mut bytes = 0;
 		for viewer in &mut self.viewers {
@@ -429,7 +449,7 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator<Item = Shape>)
 						let cluster = Cluster::new(version, *shape).await;
 						let payload = Bytes::from(vec![0; shape.frame]);
 						for track in &cluster.tracks {
-							write_group(track, &payload);
+							write_group(track, &payload, FRAMES);
 						}
 						cluster
 					})
@@ -446,7 +466,8 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator<Item = Shape>)
 						for iter in 0..iters as usize {
 							let broadcast = hosted[iter % hosted.len()];
 							let start = Instant::now();
-							let mut viewer = cluster.join(relay, std::iter::once(broadcast)).await;
+							let mut viewer = cluster.join(relay).await;
+							viewer.watch(&path(broadcast), &[TRACK]).await;
 							let bytes = read_group(&mut viewer.subscribers[0]).await;
 							elapsed += start.elapsed();
 							assert_eq!(bytes, FRAMES * shape.frame);
@@ -463,6 +484,131 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator<Item = Shape>)
 	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
+/// producer does, so relay 0 reaches each one over a peer session.
+#[derive(Clone, Copy)]
+struct Dash {
+	/// Relays in the mesh: relay 0 hosts the dash session, the rest are nodes.
+	relays: usize,
+	/// Projects with a stats broadcast on every node.
+	projects: usize,
+	/// Tracks per stats broadcast.
+	tracks: usize,
+}
+
+impl Dash {
+	fn total(&self) -> usize {
+		(self.relays - 1) * self.projects * self.tracks
+	}
+
+	fn id(&self, version: &str) -> String {
+		format!(
+			"{version}/relays={}/projects={}/tracks={}/total={}",
+			self.relays,
+			self.projects,
+			self.tracks,
+			self.total()
+		)
+	}
+}
+
+/// A mesh of stats-publishing nodes with the dash session attached.
+struct DashRoom {
+	_cluster: Cluster,
+	tracks: Vec<track::Producer>,
+	_broadcasts: Vec<broadcast::Producer>,
+	dash: Viewer,
+	payload: Bytes,
+}
+
+impl DashRoom {
+	async fn new(version: &str, shape: Dash) -> Self {
+		assert!(shape.relays > 1, "the dash needs a peer node");
+		let mut cluster = Cluster::new(
+			version,
+			Shape {
+				relays: shape.relays,
+				publishers: 0,
+				viewers: 0,
+				..Shape::BASE
+			},
+		)
+		.await;
+
+		let names: Vec<_> = (0..shape.tracks).map(|track| format!("stat{track}")).collect();
+		let mut tracks = Vec::new();
+		let mut broadcasts = Vec::new();
+		let mut paths = Vec::new();
+		for node in 1..shape.relays {
+			for project in 0..shape.projects {
+				let path = format!(".stats/{project}/node/{node}");
+				let broadcast = cluster.relays[node].publish(path.as_str(), Default::default()).unwrap();
+				for name in &names {
+					tracks.push(broadcast.create_track(name.as_str(), None).unwrap());
+				}
+				broadcasts.push(broadcast);
+				paths.push(path);
+			}
+		}
+
+		let mut dash = cluster.join(0).await;
+		for path in &paths {
+			dash.watch(path, &names).await;
+		}
+
+		let mut room = Self {
+			_cluster: cluster,
+			tracks,
+			_broadcasts: broadcasts,
+			dash,
+			payload: Bytes::from(vec![0; DASH_FRAME]),
+		};
+		room.round().await;
+		room
+	}
+
+	/// One stats tick: every track writes a single-frame group and the dash
+	/// reads it.
+	async fn round(&mut self) {
+		for track in &self.tracks {
+			write_group(track, &self.payload, 1);
+		}
+		let mut bytes = 0;
+		for subscriber in &mut self.dash.subscribers {
+			bytes += read_group(subscriber).await;
+		}
+		assert_eq!(bytes, self.tracks.len() * DASH_FRAME);
+	}
+
+	async fn measure(&mut self, iters: u64) -> Duration {
+		let start = Instant::now();
+		for _ in 0..iters {
+			self.round().await;
+		}
+		start.elapsed()
+	}
+}
+
+fn dash(c: &mut Criterion, name: &str, shapes: impl IntoIterator<Item = Dash>) {
+	let rt = runtime();
+	let shapes: Vec<_> = shapes.into_iter().collect();
+	let mut group = c.benchmark_group(format!("session_dash_{name}"));
+	group.sample_size(10);
+	for version in DASH_VERSIONS {
+		for shape in &shapes {
+			let mut room = None;
+			group.throughput(Throughput::Elements(shape.total() as u64));
+			group.bench_function(BenchmarkId::from_parameter(shape.id(version)), |b| {
+				let room = room.get_or_insert_with(|| rt.block_on(DashRoom::new(version, *shape)));
+				b.iter_custom(|iters| rt.block_on(room.measure(iters)))
+			});
+		}
+	}
+	group.finish();
+}
+
 /// Print allocations per viewer-group instead of timing, to compare across revisions.
 fn allocations() {
 	const ROUNDS: usize = 256;
@@ -565,6 +711,47 @@ fn session(c: &mut Criterion) {
 		}),
 	);
 	join(c, "relays", [1, 2, 8].map(|relays| Shape { relays, ..base }));
+
+	// Production (2026-09): ~34 nodes x >=5 projects x ~24 tracks.
+	let prod = Dash {
+		relays: 35,
+		projects: 5,
+		tracks: 24,
+	};
+	// More broadcasts at a fixed track count per broadcast.
+	dash(
+		c,
+		"projects",
+		[1, 4, 16].map(|projects| Dash {
+			relays: 5,
+			projects,
+			..prod
+		}),
+	);
+	// More tracks per broadcast at a fixed broadcast count.
+	dash(
+		c,
+		"tracks",
+		[6, 24, 96].map(|tracks| Dash {
+			relays: 5,
+			projects: 4,
+			tracks,
+		}),
+	);
+	// More peer sessions carrying the same 32 broadcasts, then the production
+	// shape.
+	dash(
+		c,
+		"relays",
+		[2, 3, 5, 9, 17, 33]
+			.map(|relays| Dash {
+				relays,
+				projects: 32 / (relays - 1),
+				..prod
+			})
+			.into_iter()
+			.chain([prod]),
+	);
 }
 
 criterion_group!(benches, session);