test(net): assert a stalled write releases the group it was serving

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

@@ -2392,6 +2392,98 @@ mod serve_group_test {
 			"rank 0 must reach the transport as send order 255: {priorities:?}",
 		);
 	}
+
+	/// A subscriber that stops reading must not pin the group it was being served.
+	///
+	/// The publisher stamps the group's cache access once per frame, immediately
+	/// before writing it, so a delivery in progress gets a full `latency_max` of
+	/// grace per frame handed out. Nothing re-stamps inside the write itself: a peer
+	/// whose flow control window stays shut for longer than the whole retention
+	/// window lets the group expire mid-stream and the stream resets with `Old`.
+	/// That is the point. Holding the group for as long as a wedged peer refuses to
+	/// read would let any subscriber pin cache indefinitely.
+	#[tokio::test(start_paused = true)]
+	async fn stalled_write_releases_the_group() {
+		let gate = kio::Producer::new(true);
+		let session = SinkSession::gated_uni(gate.consume());
+		let log = session.log.clone();
+
+		let track_priority = kio::Producer::new(0u8);
+		let subscription = Subscription {
+			session,
+			id: 0,
+			track_name: "test".into(),
+			priority: PriorityQueue::default(),
+			track_priority: track_priority.consume(),
+			track_priority_seen: 0,
+			version: Version::Lite06Wip,
+			timescale: Some(crate::Timescale::default()),
+		};
+
+		let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None);
+		let mut group = track.create_group(group::Info { sequence: 0 }).unwrap();
+		group
+			.write_frame(Timestamp::from_millis(0).unwrap(), b"first".as_slice())
+			.unwrap();
+
+		// The live edge moves on, so the served group is demoted and expirable.
+		track
+			.create_group(group::Info { sequence: 1 })
+			.unwrap()
+			.finish()
+			.unwrap();
+
+		let handle = subscription.priority.insert(Priority::new(0, 0));
+		let mut serve = std::pin::pin!(subscription.serve_group(0, handle, group.consume()));
+
+		// Write the header and the first frame, leaving the task awaiting the next.
+		assert!(futures::poll!(serve.as_mut()).is_pending());
+
+		// From here every write blocks, the way a shut flow control window does.
+		*gate.write().ok().expect("gate open") = false;
+
+		group
+			.write_frame(Timestamp::from_millis(10).unwrap(), b"second".as_slice())
+			.unwrap();
+		group
+			.write_frame(Timestamp::from_millis(20).unwrap(), b"third".as_slice())
+			.unwrap();
+		group.finish().unwrap();
+
+		// The publisher takes "second" (stamping the group) and blocks writing it.
+		assert!(futures::poll!(serve.as_mut()).is_pending());
+
+		// The write stays blocked well past the retention window while the source
+		// keeps publishing, which is what runs the expiry scan.
+		for sequence in 2..8u64 {
+			tokio::time::advance(crate::track::DEFAULT_LATENCY_MAX / 2).await;
+			track.create_group(group::Info { sequence }).unwrap().finish().unwrap();
+			assert!(futures::poll!(serve.as_mut()).is_pending());
+		}
+
+		*gate.write().ok().expect("gate open") = true;
+		let res = serve.await;
+		assert!(
+			matches!(res, Err(Error::Old)),
+			"a wedged peer must not hold an expired group open: {res:?}"
+		);
+
+		// The reason reaches the peer, so it reads as a truncated group rather than
+		// a routine cancel and it can re-request the sequence.
+		assert_eq!(log.resets(), vec![Error::Old.to_code()]);
+
+		// Only the untaken tail is lost: the frame already handed to the publisher
+		// owns its payload, so the release can't reclaim it mid-write.
+		let writes = log.writes.lock().unwrap();
+		assert!(
+			writes.windows(b"second".len()).any(|w| w == b"second"),
+			"the in-flight frame still reached the wire"
+		);
+		assert!(
+			!writes.windows(b"third".len()).any(|w| w == b"third"),
+			"the untaken tail was released with the group"
+		);
+	}
 }
 
 #[cfg(test)]

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

@@ -302,6 +302,8 @@ pub struct SinkSession {
 	/// Set by [`Self::gated_bi`]. `None` parks `open_bi` itself forever, which is all
 	/// a test driving only uni streams needs.
 	bi_gate: Option<kio::Consumer<bool>>,
+	/// Set by [`Self::gated_uni`]. `None` writes immediately.
+	uni_gate: Option<kio::Consumer<bool>>,
 	/// The ALPN to report, for a test that needs a specific negotiated version rather
 	/// than the SETUP-negotiated fallback an absent one selects.
 	protocol: Option<&'static str>,
@@ -316,6 +318,7 @@ impl SinkSession {
 		Self {
 			log,
 			bi_gate: None,
+			uni_gate: None,
 			protocol: None,
 			stats: Arc::new(Mutex::new(SinkStats::default())),
 		}
@@ -347,6 +350,22 @@ impl SinkSession {
 		Self {
 			log: Log::default(),
 			bi_gate: Some(gate),
+			uni_gate: None,
+			protocol: None,
+			stats: Arc::new(Mutex::new(SinkStats::default())),
+		}
+	}
+
+	/// Serve uni streams, holding every write while `gate` reads false.
+	///
+	/// Groups travel on uni streams, so this is how a test stalls delivery the way
+	/// QUIC flow control does: the publisher keeps its consumer and its place in the
+	/// group, but no byte reaches the wire until the gate reopens.
+	pub fn gated_uni(gate: kio::Consumer<bool>) -> Self {
+		Self {
+			log: Log::default(),
+			bi_gate: None,
+			uni_gate: Some(gate),
 			protocol: None,
 			stats: Arc::new(Mutex::new(SinkStats::default())),
 		}
@@ -381,7 +400,11 @@ impl web_transport_trait::Session for SinkSession {
 	}
 
 	async fn open_uni(&self) -> Result<Self::SendStream, Self::Error> {
-		Ok(SinkSend::new(self.log.clone()))
+		Ok(SinkSend {
+			log: self.log.clone(),
+			gate: self.uni_gate.clone(),
+			finished: false,
+		})
 	}
 
 	fn send_datagram(&self, _payload: bytes::Bytes) -> Result<(), Self::Error> {