diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 6ac125f6bf..ede10a5bbc 100644 --- a/doc/lib/rs/moq-mux.md +++ b/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 diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 2d41c024a0..9ca7faaf20 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/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 diff --git a/quest/m1/test-flakes-2/mux-live-import-clock.md b/quest/m1/test-flakes-2/mux-live-import-clock.md deleted file mode 100644 index fffbc557b8..0000000000 --- a/quest/m1/test-flakes-2/mux-live-import-clock.md +++ /dev/null @@ -1,38 +0,0 @@ -# [S] moq-mux live import tests on the paused clock - -## Goal - -The moq-mux live import restart tests (`live_import_restarts_forward_after_idle` -for ts, fmp4, and flv) advance a paused clock across their idle gap and sleep -no real time. - -## Plan - -Facts (`origin/main`, 2026-10-01): the helpers (`live_import` in -`container/{ts,fmp4}/import_test.rs`, `import` in `flv/import_test.rs`) call -`std::thread::sleep(idle)` with a 300 ms idle. The gap is measured through -`clock::Anchor`, which reads `crate::Clock::now()`, and that is -`std::time::Instant::elapsed()`, so a paused tokio clock can't move it. - -Decided (2026-10-01): `crate::Clock`'s monotonic epoch becomes a -`web_async::time::Instant`. That is tokio's clock on native, so the tests start -paused and call `tokio::time::advance`, and it takes `std::time::Instant` off the -wasm path. #4687 used the same `Instant` type, but only for the private SI -debounce after dropping its `crate::Clock`; `Clock` itself still reads -`std::time::Instant`. The wall mapping stays `SystemTime`. Rejected: injecting a `now` into the importers through -`translate_at`, which is plumbing only tests want. - -`std::time::Instant` is public at the boundary: `Clock::at(epoch, wall)` is -re-exported from `moq-mux`, and the json and binary timed writes -(`json.rs` `update`/`append`, and their `binary.rs` twins) take -`Timed<_, Instant>` through the crate-private `Clock::stamp`/`capture`. -`moq_net::Timed`'s docs say moq-mux uses `std::time::Instant`. Callers pass -`std::time::Instant` from `test_util::late_clock`, moq-cli publish, moq-hls -export, moq-audio and moq-video capture, and the catalog producer tests. - -Decided (maintainer, 2026-10-01): convert at the boundary. Public inputs stay -`std::time::Instant` and convert to the private epoch internally (native only), -so this is additive and stays on `main`. Rejected: switching the public types to -`web_async::time::Instant` as a break on `dev`. - -Public API: none. Wire: none. diff --git a/rs/moq-mux/src/clock.rs b/rs/moq-mux/src/clock.rs index ac813e0edf..367039d13e 100644 --- a/rs/moq-mux/src/clock.rs +++ b/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 { + #[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 { let unix_micros = wall @@ -51,7 +63,7 @@ fn wall_clock(wall: SystemTime) -> crate::Result { /// 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 { 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 { - 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. diff --git a/rs/moq-mux/src/container/flv/import_test.rs b/rs/moq-mux/src/container/flv/import_test.rs index 2073a6886c..c4dca482b1 100644 --- a/rs/moq-mux/src/container/flv/import_test.rs +++ b/rs/moq-mux/src/container/flv/import_test.rs @@ -734,7 +734,7 @@ async fn import(chunks: &[Vec], 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)]; diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 40c10a626d..b73b5783b8 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/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)]; diff --git a/rs/moq-mux/src/container/ts/import_test.rs b/rs/moq-mux/src/container/ts/import_test.rs index f06ed12c92..34f495b17a 100644 --- a/rs/moq-mux/src/container/ts/import_test.rs +++ b/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!(