diff --git a/quest/m1/README.md b/quest/m1/README.md index c589857182b..f503a663929 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -81,7 +81,6 @@ above them. - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges with absolute or relative ends, in either order, and replaces FETCH; merged ranges never starve a subscriber; relays fill misses by range, including over moq-transport - [Live largest location](/quest/m1/largest.md) - `largest()` replaces `latest()` in Rust and JS as a live stream of the track's largest location, known before the first group from the publisher's report - [Live TS clock validation](/quest/m1/ts-live-clock.md) - resolve the local PCR-rate failure with reproducible timing evidence and CI controls -- [TS export start-up drift](/quest/m1/ts-drift-startup.md) - a TS export never exits on start-up latency from a source within tolerance, while a far-off source still fails within seconds (#5564) - [More tests under load](/quest/m1/test-flakes-2/README.md) - the second round of load-only failures, one quest per flake, fixed at the cause - [Plan: a public model](/quest/m1/model-public-surface.md) - every crate-private model item is made public or unnecessary, so a custom publisher or subscriber builds from the public model alone - [Lazy discovery](/quest/m1/lazy-discovery/README.md) - a `consume` session requests announcements only for prefixes the origin is watching, in JS then Rust; `follow`, `broadcasts`, and the `discovery` option go away diff --git a/quest/m1/ts-drift-startup.md b/quest/m1/ts-drift-startup.md deleted file mode 100644 index 960c1e25cbb..00000000000 --- a/quest/m1/ts-drift-startup.md +++ /dev/null @@ -1,56 +0,0 @@ -# [M] A TS export never exits on start-up latency - -## Goal - -`moq export ts` never exits with "the source's clock runs ... ppm off ours" -for a source within the 30 ppm tolerance, however its latency moves while -the export starts (queueing that builds, or a backlog that drains). A source -far off real time (the 220,000 ppm case) still fails within seconds. `just -test ts --full` stops failing this way. - -## Plan - -Reported in [#5564](https://github.com/moq-dev/moq/issues/5564): 2 of 10 local -`--full` runs on `main` (3abf0161c) and 1 of 51 CI runs exited within 30 s. -The first `TS export release clock` lines read wild drift (-155315, +32045 -ppm), and passing runs log none. - -From reading the code on 5beead687 (not reproduced): `Buffer::steer` -(`rs/moq-mux/src/jitter.rs`) takes one floor every 2 s, fits an upper convex -hull, and reads both the rate and `gap` (extrapolated to `now`) from it. A -rate needs only two floors, with no minimum span or significance gate. `lost` is measured against `startup`, the gap -at the second floor. The export bails when `|slow| > 30 ppm` and `lost` -passes half the delay (450 ms at `--full`'s 900 ms). So a latency ramp after -the second floor, on a 0 ppm source, can end the export. Before #5490 -(41f3087e3) the exit also required the steered drift to reach its limit, -which made start-up noise unable to end it; `--full` (#5516) landed half an -hour after #5490. - -Decided 2026-10-15 (maintainer): - -- Reproduce first with the issue's loop, and record which floors skew the - envelope, before choosing a mechanism. -- Fix the measurement; do not add a minimum span or sample count. That keeps - the decision recorded when #5490 removed `quest/m1/slow-source.md`. -- Never widen the 30 ppm tolerance, `--delay`, or the `lost` threshold. -- Candidate mechanisms, for the implementer to pick from the evidence: - judge `lost` from observed floors instead of the envelope extrapolated to - `now`; exit only when the rate's uncertainty excludes 30 ppm; keep - `startup` floating until the rate is significant; and treat a fast - source's positive `lost` (a growing buffer) differently from loss. -- Mocked-time regressions in `jitter.rs`, each failing without the fix: a - 0 ppm source whose queueing builds 300 to 500 ms over 5 to 20 s and then - holds, the same draining, and both again with three tracks shaped like - `--full` (video sent up to 0.65 s ahead). - `a_source_far_off_real_time_fails_within_seconds` keeps passing. - -Public API: none. Wire: none. - -## Closes - -- [#5564](https://github.com/moq-dev/moq/issues/5564) - flaky: `just test ts --full` exits on the release clock's first rate estimates - -## Related - -- [Live TS clock validation](/quest/m1/ts-live-clock.md) - the grader's PCR-rate verdict on `--live`, a different layer -- [TS decoded content](/quest/m1/test-flakes-2/ts-video-content.md) - the other TS CI flake, on content loss diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 6715d535bc7..a11c6deb3d2 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -94,8 +94,9 @@ pub(crate) enum Push { /// rate is the slope of one track's upper envelope, the track whose slack keeps closest to /// it: a TS mux sends its video ahead by however full the encoder's buffer is, so the /// video's slack wanders by hundreds of ms where the audio's keeps within a few. The slack -/// now is the floors' upper envelope at that slope, so the track sent latest keeps the -/// delay, and a spell of queueing shorter than half the window moves neither. +/// now is the floors' upper envelope at that slope, kept within the floors seen, so the +/// track sent latest keeps the delay, and a spell of queueing shorter than half the window +/// moves neither. /// The backlog a network stall releases is read at once, so it gives one floor, its /// freshest frame's: stepped on decode time it would give one per step of the media it /// held, and early on, before the window fills, those would read as a source far slower @@ -103,7 +104,9 @@ pub(crate) enum Push { /// 13818-1 2.4.2.1 allows one: within [`MAX_DRIFT`] of ours, changing by at most /// [`MAX_SLEW`]. A source past that is counted, and once it has cost half the delay /// [`Buffer::push`] fails. What anchoring on frames that reached us late or early left is -/// not charged to it, while the clock pulls it shut. +/// not charged to it, while the clock pulls it shut. So a move in latency smaller than half +/// the delay never fails; early on, one past it that the floors so far trace as a line reads +/// as the rate, as a source that far off would. /// /// A zero delay holds nothing and drops nothing: each frame goes out as soon as it is /// read, ordered only among the frames read together. @@ -664,7 +667,17 @@ impl Buffer { }; // The phase rises as the source's clock runs fast, so frames arrive ever earlier. let slow = -rate; - let gap = at + self.steer.steered; + // Within the floors seen: past them the envelope is a guess at the rate, not slack any + // frame had. Once a move in latency stops, the edge it moved along runs on past the + // floors that hold after it, and would charge the rate for slack no frame lost or gained. + let (low, high) = self + .steer + .phase + .iter() + .fold((f64::INFINITY, f64::NEG_INFINITY), |(low, high), &(_, y)| { + (low.min(y), high.max(y)) + }); + let gap = at.clamp(low, high) + self.steer.steered; self.steer.source = Some(rate); // The clock pulls the startup gap shut, so it shrinks with the gap and never reopens. let startup = self.steer.startup.get_or_insert(gap); @@ -1482,15 +1495,33 @@ mod tests { rates: Vec<(f64, f64)>, } - /// Feed `secs` of 10 fps frames of tracks 1 and 2 through a 500 ms buffer from a source - /// whose clock runs `ppm` slower than ours, the frame of `track` sent at `sent` seconds of - /// our clock reaching us at `arrive(track, sent)`, or with the frame of its track before it - /// if that is later, or never if `None`, in the order they reach us, until a push fails. - async fn feed(ppm: f64, secs: f64, arrive: impl Fn(u16, f64) -> Option) -> Fed { + /// The buffer [`feed`] runs a source through: its delay and the tracks it carries. + struct Shape { + delay: Duration, + tracks: &'static [u16], + } + + /// Two tracks through a 500 ms buffer. + const PAIR: Shape = Shape { + delay: Duration::from_millis(500), + tracks: &[1, 2], + }; + + /// `just test ts --full`'s export: a video and two audio tracks through a 900 ms buffer. + const FULL: Shape = Shape { + delay: Duration::from_millis(900), + tracks: &[1, 2, 3], + }; + + /// Feed `secs` of 10 fps frames of `shape`'s tracks through its buffer from a source whose + /// clock runs `ppm` slower than ours, the frame of `track` sent at `sent` seconds of our + /// clock reaching us at `arrive(track, sent)`, or with the frame of its track before it if + /// that is later, or never if `None`, in the order they reach us, until a push fails. + async fn feed(shape: &Shape, ppm: f64, secs: f64, arrive: impl Fn(u16, f64) -> Option) -> Fed { let start = Instant::now(); - let mut buffer = Buffer::new(Duration::from_millis(500)); + let mut buffer = Buffer::new(shape.delay); let mut frames = Vec::new(); - for track in [1, 2] { + for &track in shape.tracks { let mut last = 0.0f64; for k in 0..(secs * 10.0) as u64 { let decode = Duration::from_millis(k * 100); @@ -1544,13 +1575,56 @@ mod tests { /// seconds, naming its rate, rather than drop nearly every frame for hours. #[tokio::test(start_paused = true)] async fn a_source_far_off_real_time_fails_within_seconds() { - let fed = feed(220_000.0, 3_600.0, |track, sent| (track == 1).then_some(sent)).await; + let fed = feed(&PAIR, 220_000.0, 3_600.0, |track, sent| (track == 1).then_some(sent)).await; let (at, err) = fed.err.expect("no error"); assert!(at < 10.0, "failed {at} s in"); let err = err.to_string(); assert!(err.contains("-220000.0 ppm"), "the error names the wrong rate: {err}"); } + /// A source on our clock whose latency moves by under half the delay while the export + /// starts, queueing that builds or a backlog that drains over 5 to 20 s from the first + /// frame on or once the first rates are measured, and then holds, never fails or drops a + /// frame through `shape`'s buffer. Its first track (the video) is sent `lead(sent)` ahead + /// and the rest as they decode. + async fn start_up_latency_never_fails(shape: &Shape, lead: fn(f64) -> f64) { + for draining in [false, true] { + for depth in [0.3, 0.4] { + for ramp in [5.0, 10.0, 20.0] { + for from in [0.0, 3.0, 6.0] { + let fed = feed(shape, 0.0, from + ramp + 120.0, |track, sent| { + let built = depth * ((sent - from) / ramp).clamp(0.0, 1.0); + let latency = if draining { depth - built } else { built }; + let lead = if track == 1 { lead(sent) } else { 0.0 }; + Some(sent + latency - lead) + }) + .await; + let case = match draining { + true => format!("{depth} s draining over {ramp} s from {from} s"), + false => format!("{depth} s building over {ramp} s from {from} s"), + }; + assert!(fed.err.is_none(), "{case}: {:?}", fed.err); + assert_eq!(fed.dropped, 0, "{case}"); + } + } + } + } + } + + /// Read at now, the envelope edge a move in latency ran along goes on past the floors that + /// hold after it, and would charge the rate for slack no frame lost or gained. + #[tokio::test(start_paused = true)] + async fn start_up_latency_never_fails_one_track() { + start_up_latency_never_fails(&Shape { tracks: &[1], ..FULL }, |_| 0.0).await; + } + + /// The same on `just test ts --full`'s shape: video sent up to 0.65 s ahead, wandering + /// with how full the encoder's buffer is, beside two audio tracks sent as they decode. + #[tokio::test(start_paused = true)] + async fn start_up_latency_never_fails_a_full_multiplex() { + start_up_latency_never_fails(&FULL, |sent| 0.5 + 0.15 * (sent * std::f64::consts::TAU / 7.0).sin()).await; + } + /// A stall shorter than half [`RATE_WINDOW`] drops what it made late and recovers, even /// early, when only a few floors were measured before it. #[tokio::test(start_paused = true)] @@ -1558,7 +1632,7 @@ mod tests { for at in [5.0, 9.0, 1_200.0] { for length in [3.0, 10.0, 60.0, 290.0] { let stall = at..at + length; - let fed = feed(0.0, at + length + 30.0, |track, sent| { + let fed = feed(&PAIR, 0.0, at + length + 30.0, |track, sent| { (track == 1).then_some(if stall.contains(&sent) { stall.end } else { sent }) }) .await; @@ -1577,7 +1651,7 @@ mod tests { for ppm in [-20.0, 0.0, 20.0] { for late in [0.3, 0.45] { for phase in 0..5u64 { - let fed = feed(ppm, 600.0, |track, sent| { + let fed = feed(&PAIR, ppm, 600.0, |track, sent| { let jitter = ((sent / 2.0) as u64 * 3 + phase) % 5; let late = if sent < 1.0 { late } else { 0.0 }; (track == 1).then_some(sent + late + jitter as f64 * 0.002) @@ -1599,7 +1673,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn closing_a_late_start_is_not_charged_to_the_rate() { let rise = 14_000.0..14_600.0; - let fed = feed(0.0, 16_000.0, |track, sent| { + let fed = feed(&PAIR, 0.0, 16_000.0, |track, sent| { let late = match sent { _ if sent < 1.0 => 0.45, _ if rise.contains(&sent) => 0.04, @@ -1619,7 +1693,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn the_track_sent_latest_stopping_keeps_the_clock() { for queued in [0.0, 0.4] { - let fed = feed(0.0, 300.0, |track, sent| match track { + let fed = feed(&PAIR, 0.0, 300.0, |track, sent| match track { 1 => Some(sent), _ if sent < 56.0 => Some(sent + 0.7), _ => (sent < 60.0).then_some(sent + 0.7 + queued), @@ -1637,7 +1711,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn a_wandering_track_sent_latest_does_not_move_the_rate() { for ppm in [-20.0, 0.0, 20.0] { - let fed = feed(ppm, 3_600.0, |track, sent| match track { + let fed = feed(&PAIR, ppm, 3_600.0, |track, sent| match track { 1 => Some(sent + 0.2 - 0.1 * (sent * std::f64::consts::TAU / 1_200.0).cos()), _ => Some(sent), }) @@ -1657,7 +1731,7 @@ mod tests { async fn a_spell_on_the_steady_track_keeps_the_rate_on_it() { let spell = 1_500.0..1_740.0; for ppm in [-20.0, 0.0, 20.0] { - let fed = feed(ppm, 2_400.0, |track, sent| match track { + let fed = feed(&PAIR, ppm, 2_400.0, |track, sent| match track { 1 => Some(sent + 0.2 - 0.1 * (sent * std::f64::consts::TAU / 900.0).cos()), _ => Some(sent + if spell.contains(&sent) { 0.2 } else { 0.0 }), })