fix(mux): measure live import idle gaps on the async clock

doc/lib/rs/moq-mux.md

@@ -89,8 +89,9 @@ A payload that knows when it was captured (a datagram's arrival, a sensor read)
 carries that `Instant`. The producer maps it onto the broadcast clock and writes
 it as the frame timestamp, and the entry advertises `jitter` and `delay` the way
 a media rendition does, so telemetry lagging its video shows up as `delay`. An
-instant ahead of now is refused. A device's own clock is an unrelated epoch;
-keep it in the payload.
+instant ahead of now is refused. Capture inputs stay `std::time::Instant` on
+native; browser targets refuse these native instants. A device's own clock is an
+unrelated epoch; keep it in the payload.
 
 ```rust
 telemetry.append(moq_net::Timed::from(packet).at(received_at))?;
@@ -102,7 +103,9 @@ the first frame is live on arrival, every track of the input shares that one
 mapping, and a source that restarts its timestamps continues forward after the
 real idle gap. fMP4 passthrough rewrites each fragment's `tfdt` to match. Use
 it for a live feed with its own zero; publish verbatim only when the catalog's
-clock (`Config::with_clock`) already names the source's zero.
+clock (`Config::with_clock`) already names the source's zero. The broadcast clock
+uses the async runtime's monotonic time, so native tests can pause and advance it
+with Tokio; its wall mapping stays fixed.
 
 An application running its own demuxer gets the same mapping from
 `clock::Anchor`: one per source, plus one `clock::Lane` per track. A single

quest/m1/test-flakes-2/README.md

@@ -33,7 +33,6 @@ Public API: none. Wire: none.
 - [moq-cli tests on a paused clock](/quest/m1/test-flakes-2/cli-paused-clock.md) - the fetch timeout and completion tests stop racing wall-clock budgets
 - [Subscription cut by disconnect](/quest/m1/test-flakes-2/subscription-cut.md) - a publisher disconnect never ends a subscription clean
 - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed
-- [moq-mux live import clock](/quest/m1/test-flakes-2/mux-live-import-clock.md) - the live import restart tests advance a paused clock instead of sleeping
 
 ## Related
 

rs/moq-mux/src/clock.rs

@@ -23,6 +23,18 @@ use std::time::{Duration, Instant, SystemTime};
 
 use hang::catalog::{MAX_SAFE_INTEGER, MOQ_EPOCH_UNIX_MILLIS};
 
+/// Native capture inputs use the same epoch as the async clock.
+fn monotonic(at: Instant) -> crate::Result<web_async::time::Instant> {
+	#[cfg(any(not(target_arch = "wasm32"), target_os = "wasi"))]
+	return Ok(web_async::time::Instant::from_std(at));
+
+	#[cfg(all(target_arch = "wasm32", not(target_os = "wasi")))]
+	{
+		let _ = at;
+		Err(anyhow::anyhow!("std::time::Instant inputs are unsupported in the browser").into())
+	}
+}
+
 /// The catalog clock for PTS zero at `wall`.
 fn wall_clock(wall: SystemTime) -> crate::Result<hang::catalog::Clock> {
 	let unix_micros = wall
@@ -51,7 +63,7 @@ fn wall_clock(wall: SystemTime) -> crate::Result<hang::catalog::Clock> {
 /// pinned up front and that frame's PTS converts into the mapping instead.
 #[derive(Clone, Copy, Debug)]
 pub struct Clock {
-	epoch: Instant,
+	epoch: web_async::time::Instant,
 	wall: hang::catalog::Clock,
 }
 
@@ -73,22 +85,25 @@ impl Clock {
 	/// Start a clock at the current instant, with PTS zero ten seconds earlier on both the
 	/// monotonic and the wall clock, so earlier-stamped frames of a source anchored now still map.
 	pub fn new() -> Self {
-		let (now, wall) = (Instant::now(), SystemTime::now());
+		let (now, wall) = (web_async::time::Instant::now(), SystemTime::now());
 		// Shortly after boot the monotonic clock may not reach back that far; start at now then.
 		let (epoch, wall) = match (now.checked_sub(Self::LEAD), wall.checked_sub(Self::LEAD)) {
 			(Some(epoch), Some(wall)) => (epoch, wall),
 			_ => (now, wall),
 		};
-		Self::at(epoch, wall).expect("the current wall time is representable as a broadcast clock")
+		Self {
+			epoch,
+			wall: wall_clock(wall).expect("the current wall time is representable as a broadcast clock"),
+		}
 	}
 
 	/// Start a clock at an explicit monotonic epoch and wall time.
 	///
 	/// The deterministic constructor: synthetic sources and fixtures pin both ends instead of
-	/// sampling. Refuses an unrepresentable wall.
+	/// sampling. Refuses an unrepresentable wall or a browser target without native instants.
 	pub fn at(epoch: Instant, wall: SystemTime) -> crate::Result<Self> {
 		Ok(Self {
-			epoch,
+			epoch: monotonic(epoch)?,
 			wall: wall_clock(wall)?,
 		})
 	}
@@ -103,9 +118,11 @@ impl Clock {
 	/// Map the instant a payload was captured (a datagram's arrival, a sensor read) onto this clock.
 	///
 	/// Refuses an instant ahead of now, which would claim the payload reached the transport before
-	/// it existed, and one before the clock's epoch, which no timestamp can name.
+	/// it existed, and one before the clock's epoch, which no timestamp can name. Native capture
+	/// instants are unsupported in the browser.
 	pub(crate) fn capture(&self, at: Instant) -> crate::Result<moq_net::Timestamp> {
-		if at > Instant::now() {
+		let at = monotonic(at)?;
+		if at > web_async::time::Instant::now() {
 			return Err(crate::Error::InvalidCapture);
 		}
 		let elapsed = at
@@ -456,6 +473,40 @@ mod tests {
 		));
 	}
 
+	#[tokio::test(start_paused = true)]
+	async fn paused_clock_preserves_native_boundaries() {
+		let epoch = tokio::time::Instant::now();
+		let clock = Clock::at(epoch.into_std(), moq_epoch()).unwrap();
+		let fresh = Clock::new();
+		let before = fresh.now();
+		let wall = clock.wall();
+
+		tokio::time::advance(Duration::from_secs(3)).await;
+		assert_eq!(clock.now(), us(3_000_000));
+		assert_eq!(fresh.now().as_micros() - before.as_micros(), 3_000_000);
+		assert_eq!(clock.wall(), wall);
+		assert_eq!(
+			clock.wall_clock(clock.now()).unwrap(),
+			moq_epoch() + Duration::from_secs(3)
+		);
+
+		let captured = (epoch + Duration::from_secs(2)).into_std();
+		let (timed, at) = clock.stamp(moq_net::Timed::from("payload").at(captured)).unwrap();
+		assert_eq!(timed.at, Some(us(2_000_000)));
+		assert_eq!(at, Some(us(2_000_000)));
+		let (untimed, at) = clock.stamp(moq_net::Timed::from("payload")).unwrap();
+		assert_eq!(untimed.at, Some(us(3_000_000)));
+		assert_eq!(at, None);
+		assert!(matches!(
+			clock.capture((epoch + Duration::from_secs(4)).into_std()),
+			Err(crate::Error::InvalidCapture)
+		));
+		assert!(matches!(
+			clock.capture((epoch - Duration::from_secs(1)).into_std()),
+			Err(crate::Error::InvalidCapture)
+		));
+	}
+
 	#[test]
 	fn section_advertises_wall_in_clock_timescale() {
 		// PTS zero at exactly the moq epoch advertises 0.

rs/moq-mux/src/container/flv/import_test.rs

@@ -734,7 +734,7 @@ async fn import(chunks: &[Vec<u8>], live: bool, ago: Duration, idle: Duration) -
 	let mut after = before;
 	for (i, chunk) in chunks.iter().enumerate() {
 		if i > 0 {
-			std::thread::sleep(idle);
+			tokio::time::advance(idle).await;
 		}
 		importer.decode(chunk).unwrap();
 		if i == 0 {
@@ -774,7 +774,7 @@ async fn live_import_anchors_a_late_first_frame() {
 
 /// An encoder restarting its timestamps at zero continues the broadcast forward: after the real
 /// idle gap, and with every track moving onto the one new mapping.
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn live_import_restarts_forward_after_idle() {
 	for idle in [Duration::ZERO, Duration::from_millis(300)] {
 		let input = [session(true, 5_000, 50), session(false, 0, 50)];

rs/moq-mux/src/container/fmp4/import_test.rs

@@ -1331,7 +1331,7 @@ async fn live_import(
 	let mut after = before;
 	for (i, chunk) in chunks.iter().enumerate() {
 		if i > 0 {
-			std::thread::sleep(idle);
+			tokio::time::advance(idle).await;
 		}
 		fmp4.decode(chunk).unwrap();
 		if i == 0 {
@@ -1393,7 +1393,7 @@ async fn live_import_rewrites_tfdt_onto_the_broadcast_clock() {
 
 /// A source whose decode times restart at zero continues forward after the real idle gap rather
 /// than being refused as non-monotonic, with every track moving onto one new mapping.
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn live_import_restarts_forward_after_idle() {
 	for idle in [std::time::Duration::ZERO, std::time::Duration::from_millis(300)] {
 		let input = [live_session(5_000_000, 20), live_session(0, 20)];

rs/moq-mux/src/container/ts/import_test.rs

@@ -623,7 +623,7 @@ async fn live_import(chunks: &[&[u8]], live: bool, ago: std::time::Duration, idl
 	let mut after = before;
 	for (i, chunk) in chunks.iter().enumerate() {
 		if i > 0 {
-			std::thread::sleep(idle);
+			tokio::time::advance(idle).await;
 		}
 		import.decode(chunk).unwrap();
 		if i == 0 {
@@ -663,7 +663,7 @@ async fn live_import_anchors_a_late_first_frame() {
 
 /// The same feed played twice, as when an encoder restarts its PTS, continues forward after the
 /// real idle gap instead of rewinding.
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn live_import_restarts_forward_after_idle() {
 	let data: &[u8] = include_bytes!("test_data/bbb_cbr.ts");
 	let once = live_import(&[data], false, std::time::Duration::ZERO, std::time::Duration::ZERO).await;
@@ -694,9 +694,12 @@ async fn live_import_restarts_forward_after_idle() {
 		let second = offset(|t| *t.last().unwrap());
 
 		// The second pass continues after the first plus the real idle gap, not on top of it.
+		// Translation keeps the 90 kHz source timescale, so each pass's mapping truncates to
+		// a tick, and the shift between the two can fall short by under two ticks.
 		let shift = second - first;
+		let ticks = 2 * 1_000_000u128.div_ceil(90_000) as i128;
 		assert!(
-			shift >= span + idle.as_micros() as i128,
+			shift + ticks >= span + idle.as_micros() as i128,
 			"the restart resumes after the first pass and the idle gap: {shift} < {span} + {idle:?}"
 		);
 		assert!(