test(net): cover batch-boundary and mid-batch write refusals

rs/moq-net/benches/group.rs

@@ -51,8 +51,8 @@ const PAYLOAD: usize = 64;
 
 /// Frame counts to sweep. 8192 is the largest legal group (`MAX_GROUP_FRAMES`);
 /// the 8193rd write returns `GroupTooLarge`. The top end is that full group of
-/// tiny frames.
-const COUNTS: [usize; 5] = [4, 5, 512, 2_048, 8_192];
+/// tiny frames. Counts 32 and 33 straddle a full write batch and its next flush.
+const COUNTS: [usize; 7] = [4, 5, 32, 33, 512, 2_048, 8_192];
 
 /// Cached group counts to sweep for the track-level delivery benchmarks. A track
 /// publishing one group per frame at the default 5s retention sits in the hundreds,

rs/moq-net/src/model/group.rs

@@ -4198,6 +4198,29 @@ mod test {
 	/// A track is all timed or all untimed, so a frame that doesn't match it is refused.
 	#[test]
 	fn a_mismatched_frame_is_refused() {
+		for timed in [false, true] {
+			let mut group = if timed {
+				Info { sequence: 0 }.produce()
+			} else {
+				Info { sequence: 0 }.produce_untimed()
+			};
+			let mut frames = frame::Buffer::<2>::new();
+			frames
+				.push(Frame::new(timed.then_some(Timestamp::ZERO), Bytes::from_static(b"a")))
+				.unwrap();
+			frames
+				.push(Frame::new(
+					(!timed).then_some(Timestamp::ZERO),
+					Bytes::from_static(b"b"),
+				))
+				.unwrap();
+			assert!(matches!(group.write_frames(&mut frames), Err(Error::TimestampMismatch)));
+			assert_eq!(frames.len(), 2, "a refused tail leaves the whole batch intact");
+			assert_eq!(group.frame_count(), 0, "a refused tail publishes no prefix");
+			assert!(!group.is_aborted(), "a bad timestamp does not fail the group");
+			group.finish().unwrap();
+		}
+
 		let mut untimed = Info { sequence: 0 }.produce_untimed();
 		assert!(matches!(
 			untimed.write_frame(Timestamp::ZERO, Bytes::from_static(b"x")),
@@ -5876,34 +5899,37 @@ mod test {
 		assert!(matches!(read_plain(&mut consumer), Err(Error::Lagged)));
 	}
 
-	/// A frame another copy of the group holds differently is refused after compressing it,
-	/// so the refusal fails the group: its later frames would decompress against history no
-	/// reader gets. Written whole or in a batch.
+	/// A conflict with shared content fails the whole group, including when a batch's
+	/// prefix matched. A compressed copy cannot continue with its advanced encoder.
 	#[test]
-	fn a_refused_compressed_write_fails_the_group() {
-		let cache = cache::Track::new(cache::Pool::unbounded(), Default::default(), kio::Weak::new());
-		let info = track::Info::default().with_compress(true);
-		let mut first = Producer::new(Info::from(0u64), info.clone(), cache.clone());
-		first.write_frame(None, verbose(0)).unwrap();
-		let content = first.handle.activate_content().unwrap().unwrap();
-		let copy = || {
-			let copy = Producer::new(Info::from(0u64), info.clone(), cache.clone());
-			copy.handle.share_content(content.clone());
-			copy
-		};
+	fn a_refused_shared_write_fails_the_group() {
+		for compress in [false, true] {
+			let cache = cache::Track::new(cache::Pool::unbounded(), Default::default(), kio::Weak::new());
+			let info = track::Info::default().with_compress(compress);
+			let mut first = Producer::new(Info::from(0u64), info.clone(), cache.clone());
+			first.write_frame(None, verbose(0)).unwrap();
+			first.write_frame(None, verbose(1)).unwrap();
+			let content = first.handle.activate_content().unwrap().unwrap();
+			let copy = || {
+				let copy = Producer::new(Info::from(0u64), info.clone(), cache.clone());
+				copy.handle.share_content(content.clone());
+				copy
+			};
 
-		let mut whole = copy();
-		assert!(matches!(
-			whole.write_frame(None, verbose(10)),
-			Err(Error::ProtocolViolation)
-		));
-		assert!(whole.handle.is_aborted());
+			let mut whole = copy();
+			assert!(matches!(
+				whole.write_frame(None, verbose(10)),
+				Err(Error::ProtocolViolation)
+			));
+			assert!(whole.handle.is_aborted());
 
-		let mut batch = copy();
-		let mut frames = frame::Buffer::<1>::new();
-		frames.push(Frame::new(None, verbose(10))).unwrap();
-		assert!(matches!(batch.write_frames(&mut frames), Err(Error::ProtocolViolation)));
-		assert!(batch.handle.is_aborted());
+			let mut batch = copy();
+			let mut frames = frame::Buffer::<2>::new();
+			frames.push(Frame::new(None, verbose(0))).unwrap();
+			frames.push(Frame::new(None, verbose(10))).unwrap();
+			assert!(matches!(batch.write_frames(&mut frames), Err(Error::ProtocolViolation)));
+			assert!(batch.handle.is_aborted());
+		}
 	}
 
 	/// A received slice that inflates to other than its declared size fails the read.