fix(mux): include timestamp rewind boundary details

doc/lib/rs/moq-mux.md

@@ -107,9 +107,11 @@ the clock the catalog started with. A clock set with
 real start.
 
 Group starts never go backwards. A group starting before the previous group's
-start ends the import with `TimestampRewind`, as a restarted encoder or a looping
-file wrapping to the top does, flagged MPEG-TS discontinuity or not; republish
-it as a new broadcast. Frames may still dip below the previous group's content:
+start ends the import with `TimestampRewind`, whose `timestamp` and `floor` fields
+name the refused frame and the previous group's start; the message gives both in
+microseconds. A restarted encoder or a looping file wrapping to the top does this,
+flagged MPEG-TS discontinuity or not; republish it as a new broadcast. Frames may
+still dip below the previous group's content:
 B-frames, and a keyframe overlapping the previous group's last frame. A flagged
 MPEG-TS discontinuity that jumps forward continues the broadcast.
 

quest/m1/README.md

@@ -60,7 +60,6 @@ QUIC studies there on that rule.
 - [Nightly tests on macOS and Windows](/quest/m1/nightly-platform-tests.md) - `moq-auth`, `hang`, `moq-tokio`, and `moq-native` tests run nightly where `Instant` and sockets differ, not just compile
 - [CI runner stalls](/quest/m1/ci-runner-stalls.md) - the 0.4 to 0.8 s freezes of both interop tracks on CI are attributed from a week of nightlies and fixed or told apart from playback bugs
 - [Catalog estimate rate](/quest/m1/catalog-estimate-rate.md) - a rising `jitter`/`delay` estimate republishes the catalog at most once a second, in js/publish and moq-mux
-- [Rewind detail](/quest/m1/timestamp-rewind-detail.md) - a refused timestamp rewind names the frame's timestamp and the live edge
 - [Subscriber max-delay](/quest/m1/subscriber-max-delay.md) - subscriber staleness becomes `max_delay`; publisher retention stays `max_age`
 - [Wire compatibility](/quest/m1/wire-compat.md) - a nightly run tests this checkout against the last published release for tokens, session wire, and catalog/container
 - [Same-hop importers](/quest/m1/hop-aligned-import.md) - importers fed one stream publish identical groups and timestamps, so failover between a redundant pair survives

rs/moq-mux/src/container/consumer.rs

@@ -234,8 +234,10 @@ impl<F: Container> Consumer<F> {
 					Poll::Ready(Ok(Some(Event::Frame(frame)))) => {
 						let seq = group.group.sequence;
 						let ts = frame.timestamp;
-						if self.floor.is_some_and(|floor| ts.as_micros() < floor.as_micros()) {
-							return Poll::Ready(Err(TimestampRewind.into()));
+						if let Some(floor) = self.floor
+							&& ts.as_micros() < floor.as_micros()
+						{
+							return Poll::Ready(Err(TimestampRewind { timestamp: ts, floor }.into()));
 						}
 						if self.start.is_none_or(|(start, _)| start != seq) {
 							self.start = Some((seq, ts));
@@ -474,7 +476,7 @@ impl<F: Container> Consumer<F> {
 	// A later group with a media timestamp below the latest delivered group's start is malformed:
 	// group starts never go backwards. Markers have no media timestamp, so they are not this check.
 	fn poll_malformed(&mut self, waiter: &kio::Waiter) -> Result<(), F::Error> {
-		let Some((prev_group, edge)) = self.start else {
+		let Some((prev_group, floor)) = self.start else {
 			return Ok(());
 		};
 
@@ -483,9 +485,9 @@ impl<F: Container> Consumer<F> {
 				continue;
 			}
 			if let Poll::Ready(Ok(min)) = group.poll_min_timestamp(waiter, &self.format)
-				&& min.as_micros() < edge.as_micros()
+				&& min.as_micros() < floor.as_micros()
 			{
-				return Err(TimestampRewind.into());
+				return Err(TimestampRewind { timestamp: min, floor }.into());
 			}
 		}
 
@@ -1174,7 +1176,15 @@ mod tests {
 		write_group(&mut track, 1, &[ts(0)]);
 		track.finish().unwrap();
 		let err = consumer.read().await.unwrap_err();
-		assert!(matches!(err, crate::Error::TimestampRewind(_)));
+		assert!(matches!(
+			err,
+			crate::Error::TimestampRewind(TimestampRewind { timestamp, floor })
+				if timestamp == ts(0) && floor == ts(100_000)
+		));
+		assert_eq!(
+			err.to_string(),
+			"frame timestamp 0 µs is below the previous group's start 100000 µs"
+		);
 	}
 
 	#[tokio::test]
@@ -1315,7 +1325,15 @@ mod tests {
 		track.finish().unwrap();
 		assert_eq!(consumer.read().await.unwrap().unwrap().timestamp, ts(200_000));
 		let err = consumer.read().await.unwrap_err();
-		assert!(matches!(err, crate::Error::TimestampRewind(_)));
+		assert!(matches!(
+			err,
+			crate::Error::TimestampRewind(TimestampRewind { timestamp, floor })
+				if timestamp == ts(50_000) && floor == ts(100_000)
+		));
+		assert_eq!(
+			err.to_string(),
+			"frame timestamp 50000 µs is below the previous group's start 100000 µs"
+		);
 	}
 
 	// ---- Empty payloads ----

rs/moq-mux/src/container/mod.rs

@@ -123,8 +123,17 @@ pub struct InvalidEnd;
 
 /// A group starts before the previous group did: the source restarted, which is a new broadcast.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
-#[error("frame timestamp is below the previous group's start")]
-pub struct TimestampRewind;
+#[error(
+	"frame timestamp {} µs is below the previous group's start {} µs",
+	.timestamp.as_micros(),
+	.floor.as_micros()
+)]
+pub struct TimestampRewind {
+	/// The refused frame's timestamp.
+	pub timestamp: moq_net::Timestamp,
+	/// The lowest allowed timestamp: the previous group's start.
+	pub floor: moq_net::Timestamp,
+}
 
 /// Encode and decode media frames over a moq-lite group.
 ///

rs/moq-mux/src/container/producer.rs

@@ -403,17 +403,23 @@ where
 	///
 	/// Group starts never go backwards: a keyframe below the start of the previous group, or any
 	/// frame below the start of the group before its own, returns
-	/// [`TimestampRewind`](super::TimestampRewind) without writing, the way an oversized frame is
-	/// refused. That is a restart, which is a new broadcast. Frames may still dip below the
-	/// previous group's content: B-frames, open-GOP leading pictures, and a keyframe that
-	/// overlaps the previous group's last frame.
+	/// [`TimestampRewind`](super::TimestampRewind) with the refused timestamp and that start,
+	/// without writing, the way an oversized frame is refused. That is a restart, which is a new
+	/// broadcast. Frames may still dip below the previous group's content: B-frames, open-GOP
+	/// leading pictures, and a keyframe that overlaps the previous group's last frame.
 	pub fn write(&mut self, frame: Frame) -> crate::Result<()> {
 		let floor = match frame.keyframe {
 			true => self.start,
 			false => self.floor,
 		};
-		if floor.is_some_and(|floor| timestamp_lt(frame.timestamp, floor)) {
-			return Err(super::TimestampRewind.into());
+		if let Some(floor) = floor
+			&& timestamp_lt(frame.timestamp, floor)
+		{
+			return Err(super::TimestampRewind {
+				timestamp: frame.timestamp,
+				floor,
+			}
+			.into());
 		}
 
 		// A keyframe cuts the previous group, using its timestamp as the boundary where the

rs/moq-mux/src/import/track.rs

@@ -866,6 +866,43 @@ mod tests {
 		import.finish().unwrap();
 	}
 
+	#[tokio::test(start_paused = true)]
+	async fn flac_cut_per_frame_rewind_names_timestamp_and_edge() {
+		let (broadcast, catalog) = new_broadcast();
+		let config = crate::codec::flac::Config {
+			min_block_size: 4608,
+			max_block_size: 4608,
+			min_frame_size: 0,
+			max_frame_size: 0,
+			sample_rate: 48_000,
+			channel_count: 2,
+			bits_per_sample: 24,
+			total_samples: 0,
+			md5: [0; 16],
+		};
+		let mut import = Track::audio(
+			broadcast.reserve_track("audio").unwrap(),
+			catalog.reserve(),
+			AudioInit::new(AudioFormat::Flac, config.description()),
+		)
+		.unwrap();
+
+		let floor = 1_790_802_494_898_432;
+		for timestamp in [floor - 10_000, floor] {
+			import
+				.decode(b"flac frame", Some(Timestamp::from_micros(timestamp).unwrap()))
+				.unwrap();
+			import.cut(None).unwrap();
+		}
+		let err = import
+			.decode(b"flac frame", Some(Timestamp::from_micros(floor - 1).unwrap()))
+			.unwrap_err();
+		assert!(matches!(err, crate::Error::TimestampRewind(_)), "{err:?}");
+		let message = err.to_string();
+		assert!(message.contains("1790802494898431 µs"), "{message}");
+		assert!(message.contains("1790802494898432 µs"), "{message}");
+	}
+
 	#[tokio::test(start_paused = true)]
 	async fn aac_import_attaches_audio_specific_config() {
 		let (broadcast, catalog) = new_broadcast();