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..aab7f972d42 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -88,14 +88,19 @@ 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 -/// 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. +/// 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. @@ -124,23 +129,27 @@ 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. - leads: BTreeMap, + /// How much more slack each track had than the track sent latest, in seconds, in the last + /// [`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. 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 /// 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; @@ -221,6 +230,10 @@ struct Steer { steered: f64, /// The source's clock rate off ours, as last measured: positive runs fast. source: 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 { @@ -230,6 +243,7 @@ impl Default for Steer { phase: VecDeque::new(), steered: 0.0, source: None, + startup: None, } } } @@ -240,8 +254,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 +523,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 +539,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. @@ -539,14 +553,46 @@ 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 + let (mut phase, mut at) = floor .tracks - .into_values() + .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.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))) + .expect("a floor holds a track"); + // 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 { + 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; phases.push_back((at, phase)); @@ -565,6 +611,10 @@ impl Buffer { let slow = -rate; let gap = at + 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); + *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()); @@ -582,8 +632,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 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, @@ -642,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 { @@ -1201,6 +1259,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 +1279,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 +1304,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 +1391,167 @@ 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 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, + }; + 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 { + arrived, + read: arrived, + decode: Timestamp::from_micros(decode.as_micros() as u64).unwrap(), + restart: 0, + skip: 0, + sync: true, + item: "v", + }; + match buffer.push(track, 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, |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 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 stall = at..at + length; + 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); + 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"); + } + } + } + + /// 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, |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) + }) + .await; + assert!( + fed.err.is_none(), + "{ppm} ppm, {late} s late, phase {phase}: {:?}", + fed.err + ); + } + } + } + } + + /// 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. 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() { + 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"); } }