diff --git a/quest/m1/qos/README.md b/quest/m1/qos/README.md index c85a44931f..8ab7bbd43c 100644 --- a/quest/m1/qos/README.md +++ b/quest/m1/qos/README.md @@ -44,9 +44,6 @@ on `main`. ## Required -- [Lag across a splice](/quest/m1/qos/lag-splice.md) - a route switch - neither loses pending lag weight nor keeps weighing a segment replaced - before its first frame; the line does not land until it is fixed - [Lag dashboard](/quest/m1/qos/lag-dashboard.md) - the demo stats dashboard shows viewer lag percentiles and dropped media - [Publisher timeliness](/quest/m1/qos/publisher-timeliness.md) - per diff --git a/quest/m1/qos/lag-splice.md b/quest/m1/qos/lag-splice.md deleted file mode 100644 index f7c4f583a2..0000000000 --- a/quest/m1/qos/lag-splice.md +++ /dev/null @@ -1,48 +0,0 @@ -# [S] Lag across a splice - -## Goal - -A spliced subscription's `lag` histogram stays exact across a route switch: -every byte the viewer could read is weighed once, and nothing a retired -segment produces afterwards is weighed at all, however early in its life the -segment was replaced. - -## Plan - -Decided in the 2026-10-05 audit: this quest, planned in draft -[#4381](https://github.com/moq-dev/moq/pull/4381) against the QoS line branch, -lands as a quest on `main` now that questlines are flat, and the QoS line -([#4133](https://github.com/moq-dev/moq/pull/4133)) is held until it is -fixed. The sampler exists only on the line branch, so the fix PR targets -`quest/m1/qos/README` (or lands with #4133). - -[#4298](https://github.com/moq-dev/moq/pull/4298) added the per-broadcast -viewer lag sampler (`FrontierInner::sample` in `rs/moq-net/src/stats.rs`) and -had a spliced subscription watch each segment's track -(`rs/moq-net/src/model/resume.rs`). Two Codex findings on its last round -merged unanswered, and the maintainer ruled in the 09-28 merged-PR audit that -both block the line: - -- **Pending weight is lost when the replacement has no frame yet** - ([r4113824520](https://github.com/moq-dev/moq/pull/4298#discussion_r4113824520)). - `sample` takes `state.unsampled` before it knows a source supplies `newest` - and `first`. When a capped segment folded bytes into `unsampled` and the - replacement has not produced, the `?` returns and those bytes are gone. - Keep the weight until a sample can record it, or keep the retired source's - edge alongside its weight. Still present on the line branch on 2026-10-05. -- **A segment replaced before its first frame is never unwatched** - ([r4113824522](https://github.com/moq-dev/moq/pull/4298#discussion_r4113824522)). - `unwatch` runs only when an existing segment's `end` changes. A segment - the producer replaces outright is marked `pruned` and dropped through - `retired()` without it, so if its route keeps publishing, those - unreachable bytes keep weighing this viewer's lag. Unwatch on every path - that retires or replaces a segment, not just on a cap. - -Extend the existing splice lag test (`lag_weighs_every_segment`) or add -siblings so each case fails without its fix. Rejected: two quests, since -both sit in the same splice accounting path and would conflict. - -Final lag sample (#4451) changed the same sampler; keep any weight this -quest defers in its drop-time sample. - -Public API: none. Wire: none, only the histogram's values change. diff --git a/quest/m1/quest-flat-lines.md b/quest/m1/quest-flat-lines.md index 4cfbbd4e3b..bb956c240a 100644 --- a/quest/m1/quest-flat-lines.md +++ b/quest/m1/quest-flat-lines.md @@ -57,8 +57,7 @@ Remaining: - Land each line's umbrella PR with `/quest-complete`, then delete its branch: #4403 wildcard, #4034 archive, #4039 auth, #4079 cpp, #4519 ffi-shape, - #4080 obs-moq-video, #4133 qos (held until - [Lag across a splice](/quest/m1/qos/lag-splice.md) is fixed), #4437 rs2ts, + #4080 obs-moq-video, #4133 qos, #4437 rs2ts, #4653 test-flakes-2, #4180 transport-upgrade, #4640 tstd. #4162 (audio-jitter-target) landed, and #4438 folded into rs2ts. - Fold #4255 (archive/track-timeline) into the archive branch before #4034 diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 8ff3a31084..61e9916aa0 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -486,7 +486,9 @@ impl Reader { copy.until = Some(self.newest.max(copy.track.latest())); copy.update(&self.mirrored); // A replaced route's copy may live on; what it produces from now on never - // reaches this reader, so it stops weighing lag. + // reaches this reader, so it stops weighing lag. Frames still written into a + // group the reader holds from it go unweighed too: an approximation that + // keeps per-group accounting off the frame path. if let Sub::Ready(sub) = &mut copy.sub { sub.unwatch(©.frontier); } @@ -1286,10 +1288,10 @@ mod test { fn lag_weighs_the_serving_copy() { let registry = stats::Registry::new(stats::Config::new()); let session = registry.tier(stats::Tier::default()).session("root"); - let weight = || { + let lag = || { let mut report = stats::Report::default(); registry.report(&mut report); - report.traffic[0].publisher.lag.total() + report.traffic[0].publisher.lag }; let write = |track: &track::Producer, sequence: u64, payload: &[u8]| { let mut group = track.create_group(group::Info { sequence }).unwrap(); @@ -1312,19 +1314,164 @@ mod test { // Polling resolves the copy's subscription, which names its track to the frontier. drain(&mut sub); write(&a, 0, b"a0"); - assert_eq!(weight(), 2); + assert_eq!(lag().total(), 2); // Produced before the switch but not sampled yet: still this reader's media. write(&a, 1, b"a1"); let b = copy(); routes.serve(b.consume()); + // Retire A before B has a frame. Those bytes are recorded now, at A's edge + // (500ms behind the frame the subscription opened on), not dropped and not + // held for B's lag. drain(&mut sub); + let pending = lag(); + assert_eq!(pending.total(), 4); + assert_eq!(pending.buckets()[4], 2); + write(&b, 2, b"b2"); - assert_eq!(weight(), 6); + assert_eq!(lag().total(), 6); // The replaced route keeps publishing: not this reader's media. write(&a, 3, b"a3"); + assert_eq!(lag().total(), 6); + } + + /// A copy replaced before it produces is unwatched at the switch, so what + /// that track writes afterwards does not move the histogram. + #[test] + fn lag_ignores_a_track_replaced_before_its_first_frame() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let weight = || { + let mut report = stats::Report::default(); + registry.report(&mut report); + report.traffic[0].publisher.lag.total() + }; + let write = |track: &track::Producer, sequence: u64, payload: &[u8]| { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(sequence * 500), payload).unwrap(); + group.finish().unwrap(); + }; + let drain = |sub: &mut track::Subscriber| while sub.recv_group().now_or_never().is_some() {}; + + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = logical + .consume() + .with_stats(session.egress("demo")) + .subscribe(Subscription::default().with_max_age(Duration::from_secs(10))) + .now_or_never() + .unwrap() + .unwrap(); + drain(&mut sub); + + let b = copy(); + routes.serve(b.consume()); + drain(&mut sub); + write(&a, 0, b"retired-later"); + assert_eq!(weight(), 0); + + write(&b, 1, b"live"); + assert_eq!(weight(), 4); + } + + /// A replaced copy stops counting at the switch, even while a group read from it is + /// still out: frames written into that group afterwards go unweighed, as do its + /// later groups. What it produced before the switch still counts, at its edge. + #[test] + fn lag_stops_weighing_a_replaced_copy_at_the_switch() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let lag = || { + let mut report = stats::Report::default(); + registry.report(&mut report); + report.traffic[0].publisher.lag + }; + let write = |track: &track::Producer, sequence: u64, payload: &[u8]| { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(sequence * 500), payload).unwrap(); + group.finish().unwrap(); + }; + let drain = |sub: &mut track::Subscriber| while sub.recv_group().now_or_never().is_some() {}; + + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = logical + .consume() + .with_stats(session.egress("demo")) + .subscribe(Subscription::default().with_max_age(Duration::from_secs(10))) + .now_or_never() + .unwrap() + .unwrap(); + drain(&mut sub); + let mut a0 = a.create_group(group::Info { sequence: 0 }).unwrap(); + a0.write_frame(ts(0), b"aa").unwrap(); + let held = recv(&mut sub); + assert_eq!(lag().total(), 2); + + // Written before the switch but not sampled yet: recorded at A's edge, 300ms in. + a0.write_frame(ts(300), b"bb").unwrap(); + routes.serve(copy().consume()); + assert!(sub.recv_group().now_or_never().is_none()); + let switched = lag(); + assert_eq!(switched.total(), 4); + assert_eq!(switched.buckets()[3], 2); + + a0.write_frame(ts(400), b"cccc").unwrap(); + write(&a, 1, b"dddddd"); + assert_eq!(lag().total(), 4); + drop(held); + } + + /// A route that comes back while a group from its old copy is still out is weighed + /// again, and dropping the old copy does not silence it. + #[test] + fn lag_weighs_a_route_that_comes_back() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let weight = || { + let mut report = stats::Report::default(); + registry.report(&mut report); + report.traffic[0].publisher.lag.total() + }; + let write = |track: &track::Producer, sequence: u64, payload: &[u8]| { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(sequence * 500), payload).unwrap(); + group.finish().unwrap(); + }; + let drain = |sub: &mut track::Subscriber| while sub.recv_group().now_or_never().is_some() {}; + + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = logical + .consume() + .with_stats(session.egress("demo")) + .subscribe(Subscription::default().with_max_age(Duration::from_secs(10))) + .now_or_never() + .unwrap() + .unwrap(); + drain(&mut sub); + write(&a, 0, b"aa"); + let held = recv(&mut sub); + assert_eq!(weight(), 2); + + routes.serve(copy().consume()); + drain(&mut sub); + routes.serve(a.consume()); + drain(&mut sub); + write(&a, 1, b"bbbb"); assert_eq!(weight(), 6); + + drop(held); + drain(&mut sub); + write(&a, 2, b"cccccc"); + assert_eq!(weight(), 12); } #[test] diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index bd3328a55f..5d823cff9e 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -1824,8 +1824,18 @@ struct FrontierState { behind: Option, /// The tracks feeding the subscription: its own, or the serving route's copy. sources: Vec, - /// Bytes an unwatched source produced since the last sample, still owed to it. - unsampled: u64, + /// Retired sources. Each keeps the bytes produced since its last sample and the + /// edge those bytes are weighed at, so a sample can record them before a + /// replacement has a frame, and so they never take the replacement's lag. + folded: Vec, +} + +/// Bytes a retired source produced before it stopped feeding the subscription, +/// weighed at the edge it had then. Later bytes on that track are not in here. +struct Folded { + weight: u64, + newest: Duration, + first: Duration, } /// A track feeding a [`Frontier`], with the produced bytes already sampled. @@ -1847,7 +1857,7 @@ impl Frontier { advanced: crate::model::clock::now(), behind: None, sources: Vec::new(), - unsampled: 0, + folded: Vec::new(), }), }); counters @@ -1879,21 +1889,29 @@ impl Frontier { } /// Remove a track that no longer feeds this subscription, such as a route's copy - /// replaced by a switch, so what it produces from then on stops counting. + /// replaced by a switch. Bytes it already produced stay, at the edge it had; + /// anything it produces afterwards does not. pub(crate) fn unwatch(&self, track: &Arc) { let Some(inner) = &self.0 else { return }; let mut state = inner.state.lock().expect("stats frontier poisoned"); let production = track.production(); - let mut unsampled = 0; - state.sources.retain(|s| { - if !Arc::ptr_eq(&s.production, production) { + let mut folded = Vec::new(); + state.sources.retain(|source| { + if !Arc::ptr_eq(&source.production, production) { return true; } - // What it produced before the cap reached this reader: keep it for the next sample. - unsampled += production.bytes.load(Ordering::Relaxed).saturating_sub(s.sampled); + // Acquire pairs with `Production::record`, so the edge covers these bytes. + let bytes = production.bytes.load(Ordering::Acquire); + let weight = bytes.saturating_sub(source.sampled); + if weight > 0 + && let Some(newest) = Production::load(&production.newest) + && let Some(first) = Production::load(&production.first) + { + folded.push(Folded { weight, newest, first }); + } false }); - state.unsampled += unsampled; + state.folded.extend(folded); } /// Start tracking one group stream written toward the peer. @@ -1905,12 +1923,47 @@ impl Frontier { } } +impl FrontierState { + /// Lag of `newest` against the acknowledged frontier, including how long that + /// frontier has stood still. Does not move the stall mark; [`Self::observe`] does. + fn lag(&self, newest: Duration, now: Instant) -> Duration { + let Some(acked) = self.acked else { + return Duration::ZERO; + }; + let behind = newest.saturating_sub(acked); + if behind.is_zero() { + return Duration::ZERO; + } + // Media time alone trusts the publisher's timestamps; a frontier standing + // still on the wall clock while there is newer media is lag too. + let since = self.behind.unwrap_or(now).max(self.advanced); + behind.max(now.duration_since(since)) + } + + /// Remember whether `newest` leaves the frontier behind, for the next sample's + /// wall-clock lag. + fn observe(&mut self, newest: Duration, now: Instant) { + let Some(acked) = self.acked else { + return; + }; + if newest.saturating_sub(acked).is_zero() { + self.behind = None; + return; + } + let since = self.behind.unwrap_or(now).max(self.advanced); + self.behind = Some(since); + } +} + impl FrontierInner { - /// One sample: the lag and the bytes produced since the last one, or `None` when - /// nothing was produced (a paused source moves nothing). + /// One sample of what the live sources produced since the last one, or `None` + /// when they produced nothing (a paused source moves nothing). Retired sources + /// are recorded here as well, each at its own edge, including when no live + /// source has a frame yet and including the sample taken on drop. fn sample(&self, now: Instant) -> Option<(Duration, u64)> { let mut state = self.state.lock().expect("stats frontier poisoned"); - let mut weight = std::mem::take(&mut state.unsampled); + let folded = std::mem::take(&mut state.folded); + let mut weight = 0; let mut newest = None; let mut first: Option = None; state.sources.retain_mut(|source| { @@ -1928,24 +1981,35 @@ impl FrontierInner { } live }); - let newest = newest?; - // Opened before the track produced anything: its first frame is where the - // subscription started. - let acked = *state.acked.get_or_insert(first?); + // Opened before anything produced: the earliest frame is where it started. + if state.acked.is_none() { + let mut seed = first; + for item in &folded { + seed = Some(seed.map_or(item.first, |seed| seed.min(item.first))); + } + state.acked = seed; + } + + // A retired source's bytes are not the replacement's. Record them even when + // the replacement has no frame, and never at the replacement's lag. + for item in &folded { + let lag = state.lag(item.newest, now); + self.counters.publisher.lag.record(lag, item.weight); + } + + let Some(newest) = newest else { + if let Some(edge) = folded.iter().map(|item| item.newest).max() { + state.observe(edge, now); + } + return None; + }; + // A live edge with no first frame and no acknowledged start cannot be placed. + state.acked?; + let lag = state.lag(newest, now); // Tracked even without weight, so a pause the peer caught up in doesn't read // as a stalled frontier once production resumes. - let behind = newest.saturating_sub(acked); - let lag = if behind.is_zero() { - state.behind = None; - Duration::ZERO - } else { - // Media time alone trusts the publisher's timestamps; a frontier standing - // still on the wall clock while there is newer media is lag too. - let since = state.behind.unwrap_or(now).max(state.advanced); - state.behind = Some(since); - behind.max(now.duration_since(since)) - }; + state.observe(newest, now); (weight > 0).then_some((lag, weight)) } }