fix(net): reset a subscription that cannot encode a later group

quest/m1/README.md

@@ -91,7 +91,6 @@ blocks. The quests that gated m0 lines moved under them.
 - [JS request window](/quest/m1/js-request-window.md) - @moq/net grants MAX_REQUEST_ID as requests close, so a peer never stalls after ~21k requests
 - [Relay session limits](/quest/m1/relay-session-limits.md) - moq-relay sets per-session request limits, tighter for clients than peers, and the bindings name a refused request
 - [Churn with held subscriptions](/quest/m1/session-churn-held.md) - opening and closing a request costs the same with 1 or 1,024 held subscriptions
-- [Wide varint tests](/quest/m1/varint-test-gaps.md) - wide sequences and both varint forms are tested where they differ
 - [Export linger](/quest/m1/export-ts-linger.md) - `moq export ts --linger` fails on an export error while the broadcast is live, and lingers once it closes or is replaced
 - [Archive](/quest/m1/archive/README.md) - record selected tracks to any object_store and replay them over FETCH or derived HLS; the catalog entry and format may break in place, since no archives exist
 - [In-band auth](/quest/m1/auth/README.md) - a session tells its peer what it may publish and subscribe to, unions tokens presented in band, and fails loud on an out-of-scope publish

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

@@ -566,6 +566,37 @@ mod tests {
 		assert_eq!(wire_hops, chain(2, &[5], 1));
 	}
 
+	/// Hop 100 is 2 bytes as a QUIC varint and 1 byte as leading-ones. Sharing that
+	/// single hop from 128 announcements back costs 4 bytes in both forms (the base
+	/// is 2 bytes either way). The literal chain is 5 bytes in QUIC and 4 in
+	/// leading-ones, so QUIC would compress and leading-ones ties. A tie stays literal.
+	#[test]
+	fn a_tie_stays_literal_where_hop_100_sizes_differ() {
+		let quic = Form::Quic;
+		let leading = Form::LeadingOnes { seven: true };
+		assert_eq!(varint_size(100, quic), 2);
+		assert_eq!(varint_size(100, leading), 1);
+
+		let base = 128;
+		let compressed = |form| varint_size(base, form) + varint_size(0, form) + varint_size(1, form);
+		let literal = |form| varint_size(0, form) * 2 + varint_size(1, form) + varint_size(100, form);
+		assert!(compressed(quic) < literal(quic));
+		assert_eq!(compressed(leading), literal(leading));
+
+		let mut encoder = AnnounceEncoder::new(VERSION);
+		let mut decoder = AnnounceDecoder::default();
+		start(&mut encoder, &mut decoder, "origin", &[100]);
+		for n in 1..128 {
+			let suffix = format!("f{n}");
+			start(&mut encoder, &mut decoder, &suffix, &[1]);
+		}
+		let (_, wire_path, wire_hops) = encoder.start(Path::new("again").to_owned(), hops(&[100]));
+		assert_eq!(wire_hops, HopsRef::literal(hops(&[100])));
+		let (got_path, got_hops) = decoder.start(wire_path, wire_hops).unwrap();
+		assert_eq!(got_path.as_str(), "again");
+		assert_eq!(got_hops, hops(&[100]));
+	}
+
 	#[test]
 	fn encoder_sends_literal_when_nothing_is_shared() {
 		let mut encoder = AnnounceEncoder::new(VERSION);

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

@@ -3,7 +3,7 @@ use std::{
 	collections::{BTreeSet, HashMap},
 	ops::{Bound, ControlFlow},
 	sync::{
-		Arc,
+		Arc, Mutex,
 		atomic::{AtomicU64, AtomicUsize, Ordering},
 	},
 	task::{Context, Poll, ready},
@@ -2290,10 +2290,15 @@ struct Subscription<S: crate::transport::poll::Session> {
 ///
 /// A group is pending from the moment it is queued until its stream opens, or it gives
 /// up first (expired, or the open failed) and is never counted.
+///
+/// `unencodable` rides along because every group machine and the run loop already share
+/// this handle. A header that cannot be encoded writes nothing, so the group-stream
+/// reset is not a stream the subscriber can pin to this subscription.
 #[derive(Default)]
 struct Opens {
 	pending: AtomicU64,
 	opened: AtomicU64,
+	unencodable: Mutex<Option<Error>>,
 }
 
 impl<S: crate::transport::poll::Session> Subscription<S> {
@@ -2338,6 +2343,21 @@ impl<S: crate::transport::poll::Session> Subscription<S> {
 		self.track_priority_seen
 	}
 
+	/// Remember an encode failure the group stream could not carry.
+	///
+	/// The first one wins. The run loop resets the subscribe stream with it.
+	fn note_unencodable(&self, err: &Error) {
+		let mut slot = self.opens.unencodable.lock().unwrap();
+		if slot.is_none() {
+			*slot = Some(err.clone());
+		}
+	}
+
+	/// Take the encode failure, if a group recorded one.
+	fn take_unencodable(&self) -> Option<Error> {
+		self.opens.unencodable.lock().unwrap().take()
+	}
+
 	/// Test shim: drive one group stream like the old `serve_group`, surfacing the
 	/// error the machine otherwise swallows after aborting the stream.
 	#[cfg(test)]
@@ -2478,6 +2498,12 @@ impl<S: crate::transport::poll::Session> TrackRun<S> {
 
 		// Drive the in-flight group machines; completions just retire.
 		let _ = self.children.poll(waiter);
+		// A group whose header does not fit this peer's varint wrote nothing. The
+		// reset is on a stream the subscriber never tied to this subscription, so
+		// forward it here or the reader waits forever.
+		if let Some(err) = self.ctx.take_unencodable() {
+			return Poll::Ready(Err(err));
+		}
 
 		// The first group waits for the source to resolve where its feed starts; datagrams
 		// keep flowing meanwhile.
@@ -2757,6 +2783,10 @@ impl<S: crate::transport::poll::Session> GroupServe<S> {
 						frame_start: self.frame_start,
 					};
 					if let Err(err) = writer.buffer(&lite::DataType::Group).and_then(|()| writer.buffer(&msg)) {
+						// The header wrote nothing, so the group-stream reset has nothing to pin
+						// it to this subscription. The run loop reads the slot and resets the
+						// subscribe stream instead.
+						self.ctx.note_unencodable(&err);
 						self.state = GroupState::Done;
 						writer.abort(&err);
 						return Poll::Ready(Err(err));

rs/moq-net/tests/wide_sequence.rs

@@ -57,7 +57,68 @@ async fn wide_sequence_fails_only_its_subscription() {
 	})
 	.await
 	.expect("the wide subscription hung instead of failing");
-	assert!(result.is_err(), "lite-06 cannot carry the sequence: {result:?}");
+	// The group does not fit a QUIC varint. Encode failures have no dedicated stream
+	// code, so the subscribe stream resets with INTERNAL_ERROR and the session stays up.
+	assert!(
+		matches!(result, Err(moq_net::Error::Stream(moq_net::StreamError::Internal))),
+		"lite-06 should see an internal stream error, not {result:?}"
+	);
+
+	let mut sub = remote.track("narrow").unwrap().subscribe(None).await.unwrap();
+	let mut group = sub.recv_group().await.unwrap().unwrap();
+	assert_eq!(group.read_frame().await.unwrap().unwrap().payload, b"narrow".as_ref());
+}
+
+/// A subscription that starts inside the QUIC range and later receives a group of
+/// `1 << 62` fails only that subscription. The session and the other track continue.
+#[moq_net_sim::test]
+async fn a_later_wide_group_fails_only_that_subscription() {
+	const WIDE: u64 = 1 << 62;
+
+	let publisher = produce_origin(1);
+	let relay = produce_origin(2);
+	let client = produce_origin(3);
+
+	let broadcast = publisher.create_broadcast("bench").unwrap();
+	let wide = broadcast.create_track("wide", None).unwrap();
+	let narrow = broadcast.create_track("narrow", None).unwrap();
+	broadcast.announce(Default::default()).unwrap();
+	let mut group = wide.append_group().unwrap();
+	group.write_frame(Timestamp::ZERO, b"first".as_ref()).unwrap();
+	let mut group = narrow.append_group().unwrap();
+	group.write_frame(Timestamp::ZERO, b"narrow".as_ref()).unwrap();
+
+	let mut options = MockConnectOptions::new("moq-lite-07-wip".parse::<Version>().unwrap());
+	options.server_publish = Some(publisher.consume());
+	options.client_subscribe = Some(relay.clone());
+	let _upstream = connect_mock(options).await;
+	let mut options = MockConnectOptions::new("moq-lite-06".parse::<Version>().unwrap());
+	options.server_publish = Some(relay.consume());
+	options.client_subscribe = Some(client.clone());
+	let _downstream = connect_mock(options).await;
+
+	let consumer = client.consume();
+	consumer.routed("bench").await.unwrap();
+	let remote = consumer.request_broadcast("bench", None).await.unwrap();
+
+	let mut sub = remote.track("wide").unwrap().subscribe(None).await.unwrap();
+	let mut group = sub.recv_group().await.unwrap().unwrap();
+	assert_eq!(group.sequence, 0);
+	assert_eq!(group.read_frame().await.unwrap().unwrap().payload, b"first".as_ref());
+
+	let mut group = wide.create_group(moq_net::group::Info { sequence: WIDE }).unwrap();
+	group.write_frame(Timestamp::ZERO, b"wide".as_ref()).unwrap();
+
+	let result = moq_net_sim::timeout(Duration::from_secs(5), sub.recv_group())
+		.await
+		.expect("the wide group hung the subscription instead of failing it");
+	match result {
+		Err(moq_net::Error::Stream(moq_net::StreamError::Internal)) => {}
+		Err(err) => {
+			panic!("the later wide group should fail the subscription with an internal stream error, not {err:?}")
+		}
+		Ok(_) => panic!("the later wide group should fail the subscription with an internal stream error, got a group"),
+	}
 
 	let mut sub = remote.track("narrow").unwrap().subscribe(None).await.unwrap();
 	let mut group = sub.recv_group().await.unwrap().unwrap();