fix(net): a stalled subscriber holds one parked group per subscription

quest/m1/README.md

@@ -94,7 +94,6 @@ lands before the cut.
 - [JS request owner](/quest/m1/js-request-owner.md) - every JS IETF request stream has one owning handle with an `ended` promise and one closer
 - [Container reads drop backfill](/quest/m1/read-backfill.md) - the JS container consumer drops groups below its read cursor like Rust
 - [Rust request credit](/quest/m1/rs-request-credit.md) - a moq-net request fails at once without stream credit instead of waiting
-- [A stalled subscriber holds bounded state](/quest/m1/egress-held-state.md) - group machines and bytes a credit-withholding downstream makes a publisher hold plateau at what its credit and the latency budget allow, on lite and moq-transport
 - [qmux no-wait opens](/quest/m1/qmux-no-wait.md) - the in-tree @moq/qmux rejects an over-limit create when waitUntilAvailable is false, like Chrome
 - [Timed-out requests clean up](/quest/m1/legacy-request-timeout-cleanup.md) - on drafts 14-16, Rust and JS send PUBLISH_NAMESPACE_DONE or UNSUBSCRIBE_NAMESPACE for a timed-out request before its local reset
 - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide

rs/moq-net/benches/session.rs

@@ -16,6 +16,10 @@
 //! groups have a `_cpu` suffix. Cache-clock advancement remains outside both clocks.
 //! `session_delivery_group_cost` sweeps publishers and viewers independently for SUBSCRIBE
 //! and FETCH; its constrained cases release held streams only after a real blocked write.
+//! `session_delivery_credit` grants a viewer one uni stream at a time, swept over its
+//! subscriptions and the backlog each holds, so every group waits for a stream grant. A
+//! publisher that parks state per backlogged group pays for it on each grant, which shows
+//! as a slope over the backlog.
 //! `SESSION_EXPIRY_ONLY=1` uses the standalone 30s-expiry pool and reports groups
 //! written plus charged cache bytes every 10s during long profiling runs. Each round
 //! advances the cache clock by a second, so the expiry window spans 30 rounds.
@@ -111,6 +115,8 @@ struct Shape {
 	history: usize,
 	/// Bytes queued before a mock stream applies backpressure.
 	window: usize,
+	/// Group streams a viewer's relay may have open toward it at once.
+	credit: Option<usize>,
 	/// Read each group via a fresh FETCH rather than a live subscription.
 	fetch: bool,
 }
@@ -130,6 +136,7 @@ impl Shape {
 		burst: 1,
 		history: 1,
 		window: 1024 * 1024,
+		credit: None,
 		fetch: false,
 	};
 
@@ -154,6 +161,11 @@ impl Shape {
 	}
 
 	fn id(&self, version: &str) -> String {
+		// The credit sweep names only what it varies: the full form passes Criterion's
+		// 100-character title limit, which truncates and renumbers it.
+		if let Some(credit) = self.credit {
+			return format!("{version}/watch={}/burst={}/credit={credit}", self.watch, self.burst);
+		}
 		format!(
 			"{version}/relays={}/publishers={}/broadcasts={}/viewers={}/watch={}/frame={}{}{}{}{}{}{}",
 			self.relays,
@@ -305,6 +317,7 @@ fn path(broadcast: usize) -> String {
 struct Cluster {
 	version: Version,
 	window: usize,
+	credit: Option<usize>,
 	relays: Vec<origin::Producer>,
 	/// One track per broadcast, indexed like [`path`].
 	tracks: Vec<track::Producer>,
@@ -321,6 +334,7 @@ impl Cluster {
 		let mut this = Self {
 			version: version.parse().unwrap(),
 			window: shape.window,
+			credit: shape.credit,
 			relays: Vec::new(),
 			tracks: Vec::new(),
 			_broadcasts: Vec::new(),
@@ -393,6 +407,9 @@ impl Cluster {
 		let (origin, pool) = self.origin(CLIENT_CACHE);
 		let mut options = MockConnectOptions::new(self.version);
 		options.transport.stream_buffer = self.window;
+		// Draft 22, the one IETF version here, keeps its SETUP stream open beside the groups.
+		options.transport.max_concurrent_uni_streams =
+			self.credit.map(|credit| credit + usize::from(self.version.is_ietf()));
 		// 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());
@@ -1249,6 +1266,22 @@ fn session(c: &mut Criterion) {
 			..base
 		}),
 	);
+	// The same backlog behind one uni stream grant at a time.
+	delivery(
+		c,
+		"credit",
+		[1, 16, 64].into_iter().flat_map(|watch| {
+			[1, 16, 64].map(move |burst| Shape {
+				publishers: 1,
+				broadcasts: 64,
+				viewers: 1,
+				watch,
+				burst,
+				credit: Some(1),
+				..base
+			})
+		}),
+	);
 	delivery(
 		c,
 		"relays",