From 7578584faabbc811026e5b64a014b702e7bcdba9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 13 Oct 2026 16:51:43 -0700 Subject: [PATCH 1/5] quest(m1): claim slow-source Co-Authored-By: Claude Opus 5.5 From ef5f5364749439838a3bd46ab558e698d7d24b2a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 13 Oct 2026 19:12:40 -0700 Subject: [PATCH 2/5] fix(moq-mux): a source the output cannot follow fails the export promptly The jitter buffer failed a source past the 13818-1 drift limit only once its steered clock sat at the limit, which takes about three hours under the slew limit, so a grossly slow source (an importer 22% slow) dropped nearly every frame instead. It now fails once the measured rate is past the limit and half the delay is gone. Steering steps run on our clock rather than decode time, so the backlog a brief network stall releases gives one floor instead of one per step of the media it held; early in a run those floors read as a source far slower than ours and would have tripped the new rule. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 5 +- quest/m1/README.md | 1 - quest/m1/slow-source.md | 48 -------- rs/moq-mux/src/container/flv/export.rs | 3 +- rs/moq-mux/src/container/mkv/export.rs | 3 +- rs/moq-mux/src/container/ts/export.rs | 4 +- .../src/container/ts/export_timing_test.rs | 17 +++ rs/moq-mux/src/jitter.rs | 114 ++++++++++++++++-- 8 files changed, 129 insertions(+), 66 deletions(-) delete mode 100644 quest/m1/slow-source.md diff --git a/doc/bin/cli.md b/doc/bin/cli.md index d2ffaa38f0b..dee12a8e2a6 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -332,8 +332,9 @@ keyframe. The delay also bounds how stale a group may get, so `--delay 0` writes frame as soon as it is read and drops none as late, but still skips a stalled group once a newer one is ready. FLV and MKV reject their former `--max-delay` spelling with a migration -message. TS output's PCR follows the source's clock within what ISO/IEC -13818-1 allows. +message. The output follows the source's clock within what ISO/IEC 13818-1 +allows a TS PCR. A source further off fails the export, naming its rate, once +it has used half the delay; a brief network stall only drops what it made late. `export ts --passthrough` reads the track an `import ts --passthrough` published and writes its packets back out unchanged, from its first group on, each run diff --git a/quest/m1/README.md b/quest/m1/README.md index 0aa9ab6b4c0..a027754daaa 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -87,7 +87,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 -- [Slow source fails](/quest/m1/slow-source.md) - an export whose source runs further off real time than the output may follow fails promptly naming the rate, instead of dropping frames for hours - [Same-instance TS resume](/quest/m1/ts-same-instance-resume.md) - prove valid continuation and fix the unexpected final exporter failure without weakening identity - [Wire compat media](/quest/m1/wire-compat-media.md) - wire-compat always runs same-version media cells through the compat reader, so planned breaks can't hide its regressions - [Binding panic supervision](/quest/m1/ffi-panic-supervision.md) - a panic in any moq-c or moq-ffi background task reaches the host as a panic, never a lost callback or Cancel diff --git a/quest/m1/slow-source.md b/quest/m1/slow-source.md deleted file mode 100644 index a95251d4bc2..00000000000 --- a/quest/m1/slow-source.md +++ /dev/null @@ -1,48 +0,0 @@ -# [S] A source the output cannot follow fails the export promptly - -## Goal - -An export whose source arrives further off real time than the output's -clock may follow ends with an error naming the measured rate, once it has -used half its delay. It no longer drops nearly every frame for hours with -only per-frame "missed its deadline" warnings. A short stall, such as a -reconnect, still drops late frames and recovers. - -## Plan - -Found on #5435: before #5292, `moq import` fed a 20 s TS clip about 22% -slow on CI's debug build. The TS export measured -222,000 ppm, dropped -1,273 frames as late, and exited clean with 3.2 s captured. - -`jitter::Buffer` (`rs/moq-mux/src/jitter.rs`) already bails when the source -is out of tolerance, but only once the steered drift sits at `MAX_DRIFT`. -At `MAX_SLEW` (0.075 Hz/s of 27 MHz) the steered clock needs about 3 hours to -reach 30 ppm, so a grossly slow source never trips it. The rate itself is -known within a few `STEER` steps. - -Decided with the maintainer on 2026-10-13: - -- Fail loud, not just warn: a source the output can never follow is refused. -- The check lives in the shared jitter buffer, so TS, FLV, and MKV export - fail the same way. Audio decode is out of scope: `moq-audio` sizes its - delay with its own estimator and doesn't steer a clock. -- The rule is the existing one minus "the steered clock is at its limit": - the measured rate is past `MAX_DRIFT` and half the delay is gone. No new - constant. Rejected: a run of a delay's worth of consecutive drops, a new - heuristic beside the rate the buffer already measures. -- Prove with mocked time that a stall shorter than half `RATE_WINDOW` (the - envelope's own bound) doesn't trip it, including an early stall when only - a few floors exist. If the envelope can't tell the two apart that early, - fix the measurement rather than adding a minimum span. -- Regressions: a source 22% slow fails within a few seconds of decode time - with the rate in the error; one 400 ppm off (the existing counted case) - fails once half the delay is gone; sources within 30 ppm never fail. -- Update any doc or comment that says late frames only drop, inline. - -Public API: none; an export that silently degraded now returns an error. -Wire: none. - -## Related - -- [Live TS clock validation](/quest/m1/ts-live-clock.md) - a local run measured -1382 ppm and exited clean; with this rule it fails loudly, which is evidence for that quest, not a blocker -- [#5435](https://github.com/moq-dev/moq/pull/5435) - the duration-fidelity investigation that found this diff --git a/rs/moq-mux/src/container/flv/export.rs b/rs/moq-mux/src/container/flv/export.rs index 9a2a00f22e8..04542ec6249 100644 --- a/rs/moq-mux/src/container/flv/export.rs +++ b/rs/moq-mux/src/container/flv/export.rs @@ -209,7 +209,8 @@ impl Export { /// Release frames a fixed delay after their decode time, dropping late frames. /// - /// The same delay bounds each source's staleness. Zero (the default) releases each frame as + /// A source whose clock runs further off ours than the output's may follow fails the export + /// once it has used half the delay. The same delay bounds each source's staleness. Zero (the default) releases each frame as /// soon as it is read and drops none as late, but a stalled group is still skipped once a /// newer one is ready. pub fn with_delay(mut self, delay: std::time::Duration) -> Self { diff --git a/rs/moq-mux/src/container/mkv/export.rs b/rs/moq-mux/src/container/mkv/export.rs index 3ae47ea0603..36176b1e8d0 100644 --- a/rs/moq-mux/src/container/mkv/export.rs +++ b/rs/moq-mux/src/container/mkv/export.rs @@ -218,7 +218,8 @@ impl Export { /// Release frames a fixed delay after their decode time, dropping late frames. /// - /// The same delay bounds each source's staleness. Zero (the default) releases each frame as + /// A source whose clock runs further off ours than the output's may follow fails the export + /// once it has used half the delay. The same delay bounds each source's staleness. Zero (the default) releases each frame as /// soon as it is read and drops none as late, but a stalled group is still skipped once a /// newer one is ready. pub fn with_delay(mut self, delay: Duration) -> Self { diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 613e99ddbce..3faf3a7aa8a 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -885,7 +885,9 @@ impl Export { /// joiner runs at the delay from its first output. The output then keeps the source's /// pace and muxes every track in `(DTS, PID)` order whatever the arrival skew between /// them. A frame that arrives after its deadline is dropped, and a video track that - /// dropped one resumes at its next keyframe. A stalled group is skipped once it falls + /// dropped one resumes at its next keyframe. A source whose clock runs further off ours than + /// 13818-1 lets the PCR follow fails the export once it has used half the delay, where a + /// brief network stall only drops what it made late. A stalled group is skipped once it falls /// half the delay behind the newest content (see [`Consumer`](crate::container::Consumer)), /// so the group after it still arrives with the other half to spare. /// diff --git a/rs/moq-mux/src/container/ts/export_timing_test.rs b/rs/moq-mux/src/container/ts/export_timing_test.rs index 79935811062..33071b3cbae 100644 --- a/rs/moq-mux/src/container/ts/export_timing_test.rs +++ b/rs/moq-mux/src/container/ts/export_timing_test.rs @@ -736,6 +736,23 @@ async fn a_source_past_the_limit_is_counted() { assert!(stats.out_of_tolerance > 0, "not counted"); } +/// The same source fails the export once it has used half the delay, about ten minutes in, +/// rather than drop nearly every frame from then on. +#[tokio::test(start_paused = true)] +#[should_panic(expected = "past the 30 ppm the output's may follow")] +async fn a_source_past_the_limit_fails_once_half_the_delay_is_gone() { + on_a_scaled_clock(1.0004, 900_000_000 / VIDEO_US).await; +} + +/// A source 22% slow, as an importer starved of CPU, fails the export within seconds. +#[tokio::test(start_paused = true)] +#[should_panic(expected = "the source's clock runs -220")] +async fn a_source_far_off_real_time_fails_within_seconds() { + let mut live = Live::new(1.0 / 1.22, 0).await; + let mut leg = live.join(Duration::ZERO).await; + live.run(10_000_000 / VIDEO_US, &mut [&mut leg], |_| false).await; +} + // A 1+1 pair is two receivers of one broadcast whose outputs a seamless switch compares packet // for packet. They join at different times and read with different cadences, and from the moment // they overlap they must render the same bytes. The continuity counters are the known exception diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index ce998d2cdfe..3b4018b647b 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -88,11 +88,14 @@ pub(crate) enum Push { /// The clock follows the source's: a source clock running slower than ours would make /// every frame late in the end, and a faster one would hold more and more. So the clock /// measures the source's rate and runs its decode timeline at it, holding the freshest -/// frames of the track sent latest a delay. Each [`STEER`] of decode time gives a floor, the +/// frames of the track sent latest a delay. Each [`STEER`] of our time gives a floor, the /// most slack a frame of that track arrived with (the least across tracks of each one's /// most), which queueing and retransmission only ever lower; the upper envelope of /// the floors over [`RATE_WINDOW`] gives the rate (its slope) and the slack now, so a -/// spell of queueing shorter than half the window moves neither. The decode timeline is +/// 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 than ours. The decode timeline is /// the output's system clock, so it keeps to what ISO/IEC 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 used half the delay [`Buffer::push`] fails. @@ -135,7 +138,7 @@ pub(crate) struct Buffer { out_of_tolerance: u64, } -/// How often the clock is steered, in decode time: long enough that the floor covers a +/// How often the clock is steered, in our time: long enough that the floor covers a /// group's worth of frames sent ahead by different amounts. const STEER: Duration = Duration::from_secs(2); /// How far back the floors are kept, in decode time: a spell of queueing shorter than half @@ -240,8 +243,8 @@ struct Floor { /// Per track, its slack past the delay, less how far the drift has moved its deadline, and /// when it decodes, in seconds. tracks: BTreeMap, - /// When the step's first frame decodes. - began: Timestamp, + /// When the step's first frame was read. + began: Instant, } struct Track { @@ -509,7 +512,7 @@ impl Buffer { arrived.max(self.horizon.unwrap_or(arrived)) + self.delay } - /// Account for the slack a frame arrived with, and every [`STEER`] of decode time step the + /// Account for the slack a frame arrived with, and every [`STEER`] of our time step the /// clock's drift toward the source's rate plus a pull back to the delay, as far as /// [`MAX_SLEW`] allows. fn steer(&mut self, key: &K, deadline: Option, arrival: &Arrival) -> anyhow::Result<()> { @@ -525,13 +528,13 @@ impl Buffer { let phase = slack - delay - (self.steer.steered + elapsed * clock.drift); let floor = self.steer.floor.get_or_insert_with(|| Floor { tracks: BTreeMap::new(), - began: arrival.decode, + began: arrival.read, }); let best = floor.tracks.entry(key.clone()).or_insert((phase, now)); if phase > best.0 { *best = (phase, now); } - if arrival.decode.as_nanos().saturating_sub(floor.began.as_nanos()) < STEER.as_nanos() { + if arrival.read.saturating_duration_since(floor.began) < STEER { return Ok(()); } // Re-anchor where this frame decodes, so the new drift starts from there. @@ -582,8 +585,8 @@ impl Buffer { // To a hundredth of a ppm, so a source right at the limit is not out by rounding. if (slow.abs() * 1e8).round() > (MAX_DRIFT * 1e8).round() { self.out_of_tolerance += 1; - // At the limit and still losing ground, with half the delay gone. - if drift == MAX_DRIFT.copysign(slow) && gap * slow.signum() < -delay / 2.0 { + // The clock can never catch up, so fail once half the delay is gone. + if gap * slow.signum() < -delay / 2.0 { anyhow::bail!( "the source's clock runs {:+.1} ppm off ours, past the {:.0} ppm the output's may follow", rate * 1e6, @@ -1201,6 +1204,8 @@ mod tests { clock: f64, clock_before: f64, out_of_tolerance: u64, + /// How far off the delay the frame the run failed on would have been held, in seconds. + off: f64, } /// Feed `hours` of 2 fps frames from a source whose clock runs `ppm` slower than ours @@ -1219,6 +1224,7 @@ mod tests { clock: 0.0, clock_before: 0.0, out_of_tolerance: 0, + off: 0.0, }; let mut last: Option<(f64, Timestamp)> = None; let mut stepped = false; @@ -1243,9 +1249,13 @@ mod tests { sync: true, item: "v", }; + let held = buffer + .at(frame.decode) + .map(|deadline| since(arrived, deadline) as f64 / 1e9); match buffer.push(1, frame) { Ok(push) => assert_eq!(push, Push::Queued, "{ppm} ppm: frame {k} was late"), Err(err) => { + run.off = held.expect("a clock") - delay.as_secs_f64(); run.err = Some(err); break; } @@ -1326,13 +1336,93 @@ mod tests { } /// A source further off than the clock may follow is counted, and fails once it has used - /// half the delay, before any frame goes late. + /// half the delay, before any frame goes late, though the clock has not reached its limit. #[tokio::test(start_paused = true)] async fn a_source_past_the_drift_limit_fails() { - for ppm in [40.0, -40.0] { + for ppm in [40.0, -40.0, 400.0, -400.0] { let run = drift(ppm, 6.0, 6.0, |_| Duration::ZERO).await; assert!(run.err.is_some(), "{ppm} ppm: no error"); assert!(run.out_of_tolerance > 0, "{ppm} ppm: not counted"); + assert!( + (0.25..0.26).contains(&run.off.abs()), + "{ppm} ppm: failed {} s off the delay", + run.off + ); + } + } + + /// What [`feed`] saw. + struct Fed { + /// The decode time a push failed at, in seconds, and its error. + err: Option<(f64, anyhow::Error)>, + dropped: u64, + /// What the last push did. + last: Option, + } + + /// Feed `secs` of 10 fps frames through a 500 ms buffer from a source whose clock runs + /// `ppm` slower than ours, the frames it sends during `stall` (seconds of our clock) + /// reaching us together at its end, until a push fails. + async fn feed(ppm: f64, secs: f64, stall: std::ops::Range) -> Fed { + let start = Instant::now(); + let mut buffer = Buffer::new(Duration::from_millis(500)); + let mut fed = Fed { + err: None, + dropped: 0, + last: None, + }; + for k in 0..(secs * 10.0) as u64 { + let decode = Duration::from_millis(k * 100); + let mut sent = decode.as_secs_f64() * (1.0 + ppm * 1e-6); + if stall.contains(&sent) { + sent = stall.end; + } + let arrived = start + Duration::from_secs_f64(sent); + tokio::time::advance(arrived.saturating_duration_since(Instant::now())).await; + let frame = Arrival { + arrived, + read: arrived, + decode: Timestamp::from_micros(decode.as_micros() as u64).unwrap(), + restart: 0, + skip: 0, + sync: true, + item: "v", + }; + match buffer.push(1, frame) { + Ok(push) => fed.last = Some(push), + Err(err) => { + fed.err = Some((decode.as_secs_f64(), err)); + break; + } + } + due(&mut buffer); + } + fed.dropped = buffer.dropped(); + fed + } + + /// A source far slower than real time, as an importer starved of CPU, fails within a few + /// 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, 0.0..0.0).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 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)] + async fn a_stall_drops_late_frames_and_recovers() { + for at in [5.0, 9.0, 1_200.0] { + for length in [3.0, 10.0, 60.0, 290.0] { + let fed = feed(0.0, at + length + 30.0, at..at + length).await; + assert!(fed.err.is_none(), "{length} s stall at {at} s: {:?}", fed.err); + assert!(fed.dropped > 0, "{length} s stall at {at} s: nothing dropped"); + assert_eq!(fed.last, Some(Push::Queued), "{length} s stall at {at} s"); + } } } From 9aafefdb433b3559329e209528f76813867a0d31 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 13 Oct 2026 19:58:30 -0700 Subject: [PATCH 3/5] fix(moq-mux): charge the rate only what it lost since it was measured TS validation failed a clean live source at its first rate measurement (+483 and +80 ppm, 4 s in). The clock had been anchored on frames that reached the export late at startup, so the slack sat 250 ms past the delay, and two floors a few ms apart over 2 s read as a rate far past the limit. Neither was the rate costing the delay. The failure now counts the gap the rate opened since it was first measured, so an anchoring offset, which the clock pulls shut, is not charged to it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/jitter.rs | 59 ++++++++++++++++++++++++++++++++-------- 1 file changed, 47 insertions(+), 12 deletions(-) diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 3b4018b647b..3c73b25abdd 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -98,7 +98,8 @@ pub(crate) enum Push { /// window fills, those would read as a source far slower than ours. The decode timeline is /// the output's system clock, so it keeps to what ISO/IEC 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 used half the delay [`Buffer::push`] fails. +/// counted, and once it has cost half the delay since its rate was first measured +/// [`Buffer::push`] fails. /// /// 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. @@ -224,6 +225,10 @@ struct Steer { steered: f64, /// The source's clock rate off ours, as last measured: positive runs fast. source: Option, + /// The gap between the slack and the delay when the rate was first measured, in seconds: + /// what anchoring on frames that reached us late or early left, which the clock pulls shut + /// and the rate did not cost. + offset: Option, } impl Default for Steer { @@ -233,6 +238,7 @@ impl Default for Steer { phase: VecDeque::new(), steered: 0.0, source: None, + offset: None, } } } @@ -568,6 +574,7 @@ impl Buffer { let slow = -rate; let gap = at + self.steer.steered; self.steer.source = Some(rate); + let lost = gap - *self.steer.offset.get_or_insert(gap); // Pull the gap shut over [`RESPONSE`], no faster than the slew limit can stop on it. let pull = (gap.abs() / RESPONSE).min((2.0 * MAX_SLEW * gap.abs()).sqrt()); @@ -585,8 +592,8 @@ impl Buffer { // To a hundredth of a ppm, so a source right at the limit is not out by rounding. if (slow.abs() * 1e8).round() > (MAX_DRIFT * 1e8).round() { self.out_of_tolerance += 1; - // The clock can never catch up, so fail once half the delay is gone. - if gap * slow.signum() < -delay / 2.0 { + // The clock can never catch up, so fail once the rate has cost half the delay. + if lost * slow.signum() < -delay / 2.0 { anyhow::bail!( "the source's clock runs {:+.1} ppm off ours, past the {:.0} ppm the output's may follow", rate * 1e6, @@ -1361,9 +1368,9 @@ mod tests { } /// Feed `secs` of 10 fps frames through a 500 ms buffer from a source whose clock runs - /// `ppm` slower than ours, the frames it sends during `stall` (seconds of our clock) - /// reaching us together at its end, until a push fails. - async fn feed(ppm: f64, secs: f64, stall: std::ops::Range) -> Fed { + /// `ppm` slower than ours, each frame sent at `sent` seconds of our clock reaching us at + /// `arrive(sent)`, or with the frame before it if that is later, until a push fails. + async fn feed(ppm: f64, secs: f64, arrive: impl Fn(f64) -> f64) -> Fed { let start = Instant::now(); let mut buffer = Buffer::new(Duration::from_millis(500)); let mut fed = Fed { @@ -1371,12 +1378,11 @@ mod tests { dropped: 0, last: None, }; + let mut last = 0.0f64; for k in 0..(secs * 10.0) as u64 { let decode = Duration::from_millis(k * 100); - let mut sent = decode.as_secs_f64() * (1.0 + ppm * 1e-6); - if stall.contains(&sent) { - sent = stall.end; - } + let sent = arrive(decode.as_secs_f64() * (1.0 + ppm * 1e-6)).max(last); + last = sent; let arrived = start + Duration::from_secs_f64(sent); tokio::time::advance(arrived.saturating_duration_since(Instant::now())).await; let frame = Arrival { @@ -1405,7 +1411,7 @@ 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, 0.0..0.0).await; + let fed = feed(220_000.0, 3_600.0, |sent| sent).await; let (at, err) = fed.err.expect("no error"); assert!(at < 10.0, "failed {at} s in"); let err = err.to_string(); @@ -1418,7 +1424,12 @@ mod tests { async fn a_stall_drops_late_frames_and_recovers() { for at in [5.0, 9.0, 1_200.0] { for length in [3.0, 10.0, 60.0, 290.0] { - let fed = feed(0.0, at + length + 30.0, at..at + length).await; + let stall = at..at + length; + let fed = feed(0.0, at + length + 30.0, |sent| match stall.contains(&sent) { + true => stall.end, + false => sent, + }) + .await; assert!(fed.err.is_none(), "{length} s stall at {at} s: {:?}", fed.err); assert!(fed.dropped > 0, "{length} s stall at {at} s: nothing dropped"); assert_eq!(fed.last, Some(Push::Queued), "{length} s stall at {at} s"); @@ -1426,6 +1437,30 @@ mod tests { } } + /// A source within what the clock may follow never fails, though its first frames reached + /// us late, so the clock was anchored that much late, and each frame jitters a few ms, so + /// the first floors, a few ms apart over two seconds, measure a rate far past the limit. + #[tokio::test(start_paused = true)] + async fn a_late_start_within_the_drift_limit_never_fails() { + 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, |sent| { + let jitter = ((sent / 2.0) as u64 * 3 + phase) % 5; + let late = if sent < 1.0 { late } else { 0.0 }; + sent + late + jitter as f64 * 0.002 + }) + .await; + assert!( + fed.err.is_none(), + "{ppm} ppm, {late} s late, phase {phase}: {:?}", + fed.err + ); + } + } + } + } + /// A joiner is handed its group at once. The clock is anchored on the freshest frame, so /// what is older than the delay is dropped without counting as late, and the next group /// goes out a delay after it arrived. From 276ccd673cbd3a4c7676853a2b1e45bb778eab5c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 14 Oct 2026 09:42:35 -0700 Subject: [PATCH 4/5] fix(moq-mux): charge the rate only what it cost, not the clock's own steering The rate's loss was the gap less the gap at its first measurement. The gap includes the clock's own steering, so pulling a late start shut read as lost delay, and a later latency rise failed a healthy source hours in. The startup gap is now tracked on its own: it shrinks as the gap closes and never reopens. A track not heard in a steering step, as audio that ended before its video, is now placed by its lead below the tracks heard, so the floor no longer leaps by how much earlier the others are sent, which read as a fast source. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/jitter.rs | 141 +++++++++++++++++++++++++++------------ 1 file changed, 99 insertions(+), 42 deletions(-) diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 3c73b25abdd..2190fc0173d 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -90,16 +90,17 @@ pub(crate) enum Push { /// measures the source's rate and runs its decode timeline at it, holding the freshest /// frames of the track sent latest a delay. Each [`STEER`] of our time gives a floor, the /// most slack a frame of that track arrived with (the least across tracks of each one's -/// most), which queueing and retransmission only ever lower; the upper envelope of -/// the floors over [`RATE_WINDOW`] gives the rate (its slope) and the slack now, so 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 than ours. The decode timeline is -/// the output's system clock, so it keeps to what ISO/IEC 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 since its rate was first measured -/// [`Buffer::push`] fails. +/// most, a track not heard placed by its lead), which queueing and retransmission only ever +/// lower; the upper envelope of the floors over [`RATE_WINDOW`] gives the rate (its slope) +/// and the slack now, so 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 +/// than ours. The decode timeline is the output's system clock, so it keeps to what ISO/IEC +/// 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. /// /// 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. @@ -128,10 +129,11 @@ pub(crate) struct Buffer { released: bool, /// What the clock is steered on, since it last started. steer: Steer, - /// Each track's slack in the last steering step, in seconds, against the others'. How far - /// ahead of its decode time a source sends each track belongs to its mux, not to one - /// generation, so a new generation keeps the track sent latest at the delay even before - /// that track reaches it. + /// How much more slack each track had than the track sent latest, in seconds, in the last + /// steering step that heard it. How far ahead of its decode time a source sends each track + /// belongs to its mux, not to one generation, so a new generation keeps the track sent + /// latest at the delay even before that track reaches it, and a step that does not hear + /// that track still measures against it. leads: BTreeMap, timer: Option>>, dropped: u64, @@ -225,10 +227,10 @@ struct Steer { steered: f64, /// The source's clock rate off ours, as last measured: positive runs fast. source: Option, - /// The gap between the slack and the delay when the rate was first measured, in seconds: - /// what anchoring on frames that reached us late or early left, which the clock pulls shut - /// and the rate did not cost. - offset: Option, + /// What is left of the gap between the slack and the delay that the rate's first + /// measurement found, in seconds: what anchoring on frames that reached us late or early + /// left, which the rate did not cost. + startup: Option, } impl Default for Steer { @@ -238,7 +240,7 @@ impl Default for Steer { phase: VecDeque::new(), steered: 0.0, source: None, - offset: None, + startup: None, } } } @@ -548,14 +550,24 @@ impl Buffer { return Ok(()); }; let floor = self.steer.floor.take().expect("a floor"); - for (key, (phase, _)) in &floor.tracks { - self.leads.insert(key.clone(), *phase); - } - let (phase, at) = floor + // A track not heard this step, as audio that ended before its video, may still be the + // one sent latest: its lead puts it below the track heard that was sent least ahead of + // it. Without that the floor would leap when it stopped, which reads as a source far + // faster than ours. + let lead = |key: &K| self.leads.get(key).copied().unwrap_or(0.0); + let latest = floor .tracks - .into_values() - .min_by(|a, b| a.0.total_cmp(&b.0)) + .iter() + .min_by(|a, b| lead(a.0).total_cmp(&lead(b.0))) + .map(|(key, &(phase, at))| (phase - lead(key), at)) .expect("a floor holds a track"); + let (phase, at) = floor + .tracks + .values() + .fold(latest, |floor, &track| if track.0 < floor.0 { track } else { floor }); + for (key, (track, _)) in &floor.tracks { + self.leads.insert(key.clone(), track - phase); + } self.steer.steered += elapsed * clock.drift; let phases = &mut self.steer.phase; phases.push_back((at, phase)); @@ -574,7 +586,10 @@ impl Buffer { let slow = -rate; let gap = at + self.steer.steered; self.steer.source = Some(rate); - let lost = gap - *self.steer.offset.get_or_insert(gap); + // 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); + *startup = startup.clamp(gap.min(0.0), gap.max(0.0)); + let lost = gap - *startup; // Pull the gap shut over [`RESPONSE`], no faster than the slew limit can stop on it. let pull = (gap.abs() / RESPONSE).min((2.0 * MAX_SLEW * gap.abs()).sqrt()); @@ -1367,22 +1382,32 @@ mod tests { last: Option, } - /// Feed `secs` of 10 fps frames through a 500 ms buffer from a source whose clock runs - /// `ppm` slower than ours, each frame sent at `sent` seconds of our clock reaching us at - /// `arrive(sent)`, or with the frame before it if that is later, until a push fails. - async fn feed(ppm: f64, secs: f64, arrive: impl Fn(f64) -> f64) -> Fed { + /// 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 { let start = Instant::now(); let mut buffer = Buffer::new(Duration::from_millis(500)); + let mut frames = Vec::new(); + for track in [1, 2] { + let mut last = 0.0f64; + for k in 0..(secs * 10.0) as u64 { + let decode = Duration::from_millis(k * 100); + if let Some(sent) = arrive(track, decode.as_secs_f64() * (1.0 + ppm * 1e-6)) { + last = sent.max(last); + frames.push((last, track, decode)); + } + } + } + frames.sort_by(|a, b| a.0.total_cmp(&b.0)); + buffer.expect(frames.iter().map(|(_, track, _)| *track)); let mut fed = Fed { err: None, dropped: 0, last: None, }; - let mut last = 0.0f64; - for k in 0..(secs * 10.0) as u64 { - let decode = Duration::from_millis(k * 100); - let sent = arrive(decode.as_secs_f64() * (1.0 + ppm * 1e-6)).max(last); - last = sent; + for (sent, track, decode) in frames { let arrived = start + Duration::from_secs_f64(sent); tokio::time::advance(arrived.saturating_duration_since(Instant::now())).await; let frame = Arrival { @@ -1394,7 +1419,7 @@ mod tests { sync: true, item: "v", }; - match buffer.push(1, frame) { + match buffer.push(track, frame) { Ok(push) => fed.last = Some(push), Err(err) => { fed.err = Some((decode.as_secs_f64(), err)); @@ -1411,7 +1436,7 @@ 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, |sent| sent).await; + let fed = feed(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(); @@ -1425,9 +1450,8 @@ 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, |sent| match stall.contains(&sent) { - true => stall.end, - false => sent, + let fed = feed(0.0, at + length + 30.0, |track, sent| { + (track == 1).then_some(if stall.contains(&sent) { stall.end } else { sent }) }) .await; assert!(fed.err.is_none(), "{length} s stall at {at} s: {:?}", fed.err); @@ -1445,10 +1469,10 @@ 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, |sent| { + let fed = feed(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 }; - sent + late + jitter as f64 * 0.002 + (track == 1).then_some(sent + late + jitter as f64 * 0.002) }) .await; assert!( @@ -1461,6 +1485,39 @@ mod tests { } } + /// Closing a late start is the clock's own steering, not what the rate cost: hours after + /// the clock pulled a 450 ms late start shut, a source within the limit does not fail on + /// a rise in latency long enough to move the measured rate past the limit. + #[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 late = match sent { + _ if sent < 1.0 => 0.45, + _ if rise.contains(&sent) => 0.04, + _ => 0.0, + }; + (track == 1).then_some(sent + late) + }) + .await; + assert!(fed.err.is_none(), "{:?}", fed.err); + } + + /// The track sent latest stopping, as audio sent 700 ms after its video ending first, keeps + /// the clock: each floor is still measured against that track, by how much earlier than it + /// the others are sent, rather than leap 700 ms, which would read as a source far faster + /// than ours. + #[tokio::test(start_paused = true)] + async fn the_track_sent_latest_stopping_keeps_the_clock() { + let fed = feed(0.0, 300.0, |track, sent| match track { + 1 => Some(sent), + _ => (sent < 60.0).then_some(sent + 0.7), + }) + .await; + assert!(fed.err.is_none(), "{:?}", fed.err); + assert_eq!(fed.dropped, 0); + } + /// A joiner is handed its group at once. The clock is anchored on the freshest frame, so /// what is older than the delay is dropped without counting as late, and the next group /// goes out a delay after it arrived. From 28a41f4e959019a4f9f278ffc7b8e7f9766469e5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 14 Oct 2026 10:29:12 -0700 Subject: [PATCH 5/5] fix(moq-mux): a track's lead is the median of a window of steps A lead was the last step's sample, so queueing on the track sent latest in its final step before it ended was frozen into the other tracks' leads, and the floors after it read as a slow source. Each lead is now the median of a window's worth of steps, so a spell of queueing shorter than half of it does not move it, as for the rate. The placed floor only applies when a track sent later than every track heard is not heard. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/jitter.rs | 87 ++++++++++++++++++++++++++++------------ 1 file changed, 62 insertions(+), 25 deletions(-) diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 2190fc0173d..aab7f972d42 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -130,11 +130,11 @@ pub(crate) struct Buffer { /// What the clock is steered on, since it last started. steer: Steer, /// How much more slack each track had than the track sent latest, in seconds, in the last - /// steering step that heard it. How far ahead of its decode time a source sends each track - /// belongs to its mux, not to one generation, so a new generation keeps the track sent - /// latest at the delay even before that track reaches it, and a step that does not hear - /// that track still measures against it. - leads: BTreeMap, + /// [`LEADS`] steering steps that heard it; [`Buffer::lead`] is their median. How far ahead + /// of its decode time a source sends each track belongs to its mux, not to one generation, + /// so a new generation keeps the track sent latest at the delay even before that track + /// reaches it, and a step that does not hear that track still measures against it. + leads: BTreeMap>, timer: Option>>, dropped: u64, /// Steps at which the source's clock ran further off ours than the clock may follow. @@ -147,6 +147,9 @@ const STEER: Duration = Duration::from_secs(2); /// How far back the floors are kept, in decode time: a spell of queueing shorter than half /// of it moves neither the rate nor the slack. const RATE_WINDOW: f64 = 600.0; +/// How many steering steps a track's lead is the median of: a window's worth, so a spell of +/// queueing shorter than half of it does not move the lead either. +const LEADS: usize = (RATE_WINDOW / STEER.as_secs_f64()) as usize; /// How quickly the clock pulls the slack back to the delay: it closes the gap over this /// many seconds, or slower where the slew limit could not stop it in time. const RESPONSE: f64 = 600.0; @@ -550,23 +553,45 @@ impl Buffer { return Ok(()); }; let floor = self.steer.floor.take().expect("a floor"); - // A track not heard this step, as audio that ended before its video, may still be the - // one sent latest: its lead puts it below the track heard that was sent least ahead of + let (mut phase, mut at) = floor + .tracks + .values() + .copied() + .min_by(|a, b| a.0.total_cmp(&b.0)) + .expect("a floor holds a track"); + // A track not heard this step, as audio that ended before its video, may be sent later + // than every track heard: its lead places it below the heard track sent least ahead of // it. Without that the floor would leap when it stopped, which reads as a source far // faster than ours. - let lead = |key: &K| self.leads.get(key).copied().unwrap_or(0.0); - let latest = floor + let lead = |key: &K| self.lead(key).unwrap_or(0.0); + let unheard = self + .leads + .keys() + .filter(|key| !floor.tracks.contains_key(*key)) + .map(lead) + .reduce(f64::min); + let (heard, &(heard_phase, heard_at)) = floor .tracks .iter() .min_by(|a, b| lead(a.0).total_cmp(&lead(b.0))) - .map(|(key, &(phase, at))| (phase - lead(key), at)) .expect("a floor holds a track"); - let (phase, at) = floor - .tracks - .values() - .fold(latest, |floor, &track| if track.0 < floor.0 { track } else { floor }); + // Where the floor sits among the leads. + let mut base = 0.0; + if let Some(unheard) = unheard + && unheard < lead(heard) + { + let placed = heard_phase - (lead(heard) - unheard); + if placed < phase { + (phase, at) = (placed, heard_at); + } + base = unheard + (phase - placed); + } for (key, (track, _)) in &floor.tracks { - self.leads.insert(key.clone(), track - phase); + let leads = self.leads.entry(key.clone()).or_default(); + leads.push_back(track - phase + base); + if leads.len() > LEADS { + leads.pop_front(); + } } self.steer.steered += elapsed * clock.drift; let phases = &mut self.steer.phase; @@ -667,19 +692,27 @@ impl Buffer { /// not heard from is sent, by the leads the previous generation measured: anchoring that /// much later keeps that track at the delay when it reaches the new generation. fn unheard_lag(&self, acquire: &Acquire, fresh: &K) -> Duration { - let Some(lead) = self.leads.get(fresh) else { + let Some(lead) = self.lead(fresh) else { return Duration::ZERO; }; let lag = self .expect .iter() .filter(|key| !acquire.tracks.contains_key(key)) - .filter_map(|key| self.leads.get(key)) + .filter_map(|key| self.lead(key)) .map(|other| lead - other) .fold(0.0, f64::max); Duration::from_secs_f64(lag) } + /// How much more slack a track has than the track sent latest, in seconds: the median of + /// its last [`LEADS`], so a spell of queueing does not move it. + fn lead(&self, key: &K) -> Option { + let mut leads: Vec = self.leads.get(key)?.iter().copied().collect(); + leads.sort_by(f64::total_cmp); + Some(leads[leads.len() / 2]) + } + /// When `acquire` ends: a track not yet heard from may be the one sent latest, so until /// every expected track has delivered it waits, two delays at most. fn ends(&self, acquire: &Acquire) -> Instant { @@ -1506,16 +1539,20 @@ mod tests { /// The track sent latest stopping, as audio sent 700 ms after its video ending first, keeps /// the clock: each floor is still measured against that track, by how much earlier than it /// the others are sent, rather than leap 700 ms, which would read as a source far faster - /// than ours. + /// than ours. That holds when its last frames were queued 400 ms more on their way, since + /// a spell of queueing does not move the lead. #[tokio::test(start_paused = true)] async fn the_track_sent_latest_stopping_keeps_the_clock() { - let fed = feed(0.0, 300.0, |track, sent| match track { - 1 => Some(sent), - _ => (sent < 60.0).then_some(sent + 0.7), - }) - .await; - assert!(fed.err.is_none(), "{:?}", fed.err); - assert_eq!(fed.dropped, 0); + for queued in [0.0, 0.4] { + let fed = feed(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), + }) + .await; + assert!(fed.err.is_none(), "{queued} s queued: {:?}", fed.err); + assert_eq!(fed.dropped, 0, "{queued} s queued"); + } } /// A joiner is handed its group at once. The clock is anchored on the freshest frame, so