From db852aa676e96e351fbfbad691ca865118a898b2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 15:43:03 -0700 Subject: [PATCH 1/5] chore(quest): claim quest/m1/catalog-estimates.md Quest-Claim: bb79474b6f256afd From e83ad8050232a5dff6e14ad0bae0a428f288c392 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 17:04:26 -0700 Subject: [PATCH 2/5] fix(mux): settle catalog estimates before they churn Hang republishes a measured jitter or delay rise only once it reaches the next whole frame duration above the published value: 1/framerate when the config has one, else the shortest timestamp advance measured. MSF waits until every unhinted video and audio track has closed a group after a second of media (and measured its framerate window, for unhinted video), declares its estimates once, freezes them, and skips writing a catalog identical to the last one. A finished track stops holding MSF back. Closes #7067 Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/catalog-estimates.md | 59 ---- rs/moq-mux/src/catalog/data.rs | 8 +- rs/moq-mux/src/catalog/estimate.rs | 42 +++ rs/moq-mux/src/catalog/mod.rs | 1 + rs/moq-mux/src/catalog/producer.rs | 335 +++++++++++++++++-- rs/moq-mux/src/catalog/tracks.rs | 100 +++++- rs/moq-mux/src/container/fmp4/import.rs | 21 +- rs/moq-mux/src/container/fmp4/import_test.rs | 3 + rs/moq-mux/src/container/producer.rs | 17 +- 10 files changed, 476 insertions(+), 111 deletions(-) delete mode 100644 quest/m1/catalog-estimates.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 32a58c043cd..48b514fa1cf 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -155,7 +155,6 @@ removed). - [JS FETCH timeout test](/quest/m1/js-fetch-timeout-test.md) - a JS joining FETCH still waiting on its SUBSCRIBE at the deadline is refused with TIMEOUT - [JS joins at the current group](/quest/m1/ietf-js-join.md) - a late JS moq-transport subscriber gets the group in progress whole through a joining FETCH or fill, as Rust does - [IETF fill timeout](/quest/m1/ietf-fill-timeout.md) - Rust honors native FILL_TIMEOUT to bound upstream waiting on drafts 18 and 22, alongside local content-age enforcement -- [Catalog estimates settle](/quest/m1/catalog-estimates.md) - hang quantizes jitter and delay rises, and MSF publishes once its estimates are made and then only for track changes - [IETF takeover](/quest/m1/ietf-takeover.md) - carry the shared takeover contract through the negotiated IETF epoch extension - [Identity and recovery guide](/quest/m1/identity-guide.md) - explain epochs, pinned catalogs, lazy jobs, restart and version compatibility as they exist; takeover adds its own section - [Native rung disable](/quest/m1/simulcast-rung-disable.md) - OBS stops sending the top renditions its grant cannot fund, as the Rust/FFI publishers do, and a shaped uplink through moq-ffi proves the resume diff --git a/quest/m1/catalog-estimates.md b/quest/m1/catalog-estimates.md deleted file mode 100644 index ecd245bbc1a..00000000000 --- a/quest/m1/catalog-estimates.md +++ /dev/null @@ -1,59 +0,0 @@ -# [M] Catalog estimates settle before they churn - -## Goal - -An unhinted moq-mux publisher's catalogs stop changing for small estimate -creep. Hang republishes `jitter` and `delay` only when they cross a coarse -step. The MSF catalog publishes once its estimates are made and then changes -only when the track list or a track's identity does. MSF-01 §5.3 fixes a -declared track's attributes. - -## Plan - -From [#7067](https://github.com/moq-dev/moq/issues/7067): an SDI encoder -through moq-c published 68 catalog versions in 32 s, and 16 even with every -hint from [#7068](https://github.com/moq-dev/moq/pull/7068). Of the ten that -weren't structural, eight were `delay` rising in ~8 ms steps. -[#6595](https://github.com/moq-dev/moq/pull/6595) already measures from -timestamps. `delay` still follows the cross-track write order, which tracks -runtime encode latency. It only rises, at most once per `ESTIMATE_WINDOW` -(`rs/moq-mux/src/catalog/tracks.rs:655-700`). - -Decided 2026-10-28 (maintainer): - -- **`jitter` and `delay` stay in hang.** @moq/watch, moq-audio and the TS - export use them as latency floors, and a receiver can't measure cross-track - skew itself. Hang keeps updating them in place; only a decoder identity - change publishes a new track - ([#7110](https://github.com/moq-dev/moq/pull/7110)). -- **Hang quantizes the rise.** A new `jitter` or `delay` republishes only when - it crosses the next multiple of the track's frame duration. Measurement is - unchanged. -- **MSF deduplicates.** `emit` (`rs/moq-msf` producer, `catalog/producer.rs:221-243`) - skips a group whose serialized catalog matches the previous one. -- **MSF waits for estimates, then freezes them.** MSF publishes once every - track with an unhinted estimate has closed its first group and carried - `RATE_WINDOW` (1 s, `catalog/estimate.rs:7`) of media, whichever is later. - The minimum duration matters because bitrate is measured over that window: - 20 ms audio groups would otherwise freeze it absent. Its estimated fields - then stay as first declared, and MSF republishes only for track-list or - identity changes. A track whose every estimated field is hinted publishes - at once. moq-c can hint bitrate and framerate but not jitter or delay - ([#7068](https://github.com/moq-dev/moq/pull/7068)), so its tracks wait. -- **No opt-out.** Quantizing and freezing remove the churn without new config - surface on fields players rely on. -- The HLS pointer goes away with - [time-addressed consumers](/quest/m1/time-addressing/consumers.md), so it is - out of scope here. - -Tests: a measured `delay` creeping in 1 ms steps republishes hang once per -frame-duration step. Identical MSF content publishes once. An unhinted MSF -catalog appears once the first groups close and a second of media has -passed, carries a bitrate for 20 ms audio groups, and keeps its estimates. - -Public API: none. Wire: fewer catalog groups; MSF estimate fields frozen at -first declaration. - -## Closes - -- [#7067](https://github.com/moq-dev/moq/issues/7067) - a catalog describes what a track is: keep channel measurements from churning it diff --git a/rs/moq-mux/src/catalog/data.rs b/rs/moq-mux/src/catalog/data.rs index c62b4199327..7279138e414 100644 --- a/rs/moq-mux/src/catalog/data.rs +++ b/rs/moq-mux/src/catalog/data.rs @@ -45,7 +45,7 @@ pub(crate) struct Listing { trait Owned: Send + Sync { fn name(&self) -> &str; fn clock(&self) -> crate::Clock; - fn estimate(&mut self, estimate: super::Estimate) -> crate::Result<()>; + fn estimate(&mut self, reading: super::Reading) -> crate::Result<()>; } impl> Owned for Rendition { @@ -55,8 +55,8 @@ impl> Owned for Rendition { fn clock(&self) -> crate::Clock { Rendition::clock(self) } - fn estimate(&mut self, estimate: super::Estimate) -> crate::Result<()> { - Rendition::estimate(self, estimate) + fn estimate(&mut self, reading: super::Reading) -> crate::Result<()> { + Rendition::estimate(self, reading) } } @@ -111,6 +111,6 @@ impl Listing { // media track behind it by however fast it runs. It was written no later than now. self.estimator.order(captured.min(now)); } - self.rendition.estimate(self.estimator.estimate()) + self.rendition.estimate(self.estimator.reading()) } } diff --git a/rs/moq-mux/src/catalog/estimate.rs b/rs/moq-mux/src/catalog/estimate.rs index 31d5ab74775..c4da7393249 100644 --- a/rs/moq-mux/src/catalog/estimate.rs +++ b/rs/moq-mux/src/catalog/estimate.rs @@ -190,6 +190,15 @@ impl Estimator { self.jitter.max = self.jitter.max.max(duration); } + /// Everything measured so far, with what a catalog rendition needs to decide when to publish it. + pub(crate) fn reading(&self) -> Reading { + Reading { + estimate: self.estimate(), + period: self.framerate.shortest, + framerate: self.framerate.measured, + } + } + /// Everything measured so far. /// /// Catalog-owned [`container::Producer`](crate::container::Producer) handles publish this @@ -204,6 +213,28 @@ impl Estimator { } } +/// An [`Estimate`] with the measurements that pace its publishing. +#[derive(Clone, Debug, Default)] +pub(crate) struct Reading { + pub estimate: Estimate, + /// The shortest advance between frames, which a published `jitter` or `delay` rises in whole + /// steps of when no framerate says the frame duration. + pub period: Option, + /// Whether the framerate's first window has closed. It opens on the first group boundary and + /// closes a [`RATE_WINDOW`] later. + pub framerate: bool, +} + +impl From for Reading { + fn from(estimate: Estimate) -> Self { + Self { + estimate, + period: None, + framerate: false, + } + } +} + /// A frame time as scale-free nanoseconds. /// /// Timestamps reaching an estimator can carry different timescales: a track normalizes them into @@ -307,10 +338,18 @@ struct Framerate { /// Frames written since the window opened. frames: u64, max: Option, + /// Whether a window has closed, whether or not its rate snapped. + measured: bool, + /// The shortest advance of the highest timestamp: a frame's duration, or a reordered group's + /// span of them. + shortest: Option, } impl Framerate { fn write(&mut self, ts: u128) { + if let Some(advance) = self.high.and_then(|high| elapsed(high, ts)) { + self.shortest = Some(self.shortest.map_or(advance, |shortest| shortest.min(advance))); + } self.high = Some(self.high.map_or(ts, |high| high.max(ts))); self.frames += 1; } @@ -341,6 +380,7 @@ impl Framerate { { self.max = Some(framerate); } + self.measured = true; self.open = Some(high); self.frames = 0; } @@ -349,6 +389,8 @@ impl Framerate { fn discontinuity(&mut self) { *self = Self { max: self.max, + measured: self.measured, + shortest: self.shortest, ..Self::default() }; } diff --git a/rs/moq-mux/src/catalog/mod.rs b/rs/moq-mux/src/catalog/mod.rs index 4016f61b8e9..3dde971992e 100644 --- a/rs/moq-mux/src/catalog/mod.rs +++ b/rs/moq-mux/src/catalog/mod.rs @@ -40,6 +40,7 @@ pub use consumer::Consumer; pub use data::IntoRendition; pub(crate) use data::Listing; pub use entry::Entry; +pub(crate) use estimate::Reading; pub use estimate::{Estimate, Estimator}; pub use format::*; pub use producer::{Config, Guard, Producer}; diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index e0c4a0315ce..08f7b67df48 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -53,6 +53,9 @@ struct State { /// The last revision [`Producer::fresh_track`] handed out per base name, so a successor never /// reuses a name. revisions: std::collections::HashMap, + + /// What the MSF catalog has declared, and what it still waits for. + msf: Msf, } impl State { @@ -109,6 +112,89 @@ impl Throttle { } } +/// The MSF catalog's publish state. +/// +/// MSF-01 §5.3 fixes a declared track's attributes, so the MSF catalog waits for each track's +/// estimates to be made, declares them once, and then changes only when the track list or a track's +/// identity does. The hang catalog keeps updating its estimates in place. +#[derive(Default)] +struct Msf { + /// The listed video and audio renditions whose estimates are still being made, keyed like + /// [`State::owned`]. Any enabled one holds the MSF catalog back. + unsettled: std::collections::BTreeSet<(std::any::TypeId, String)>, + /// The estimates each MSF track was first declared with, by namespace and name. + declared: HashMap<(Option, String), Declared>, + /// The last MSF catalog written, so an identical one isn't written again. + last: Option, +} + +impl Msf { + /// The MSF catalog to write for `catalog`, or `None` while a track's estimates are still being + /// made or when it matches the last one written. + fn render(&mut self, catalog: &Catalog, base: &moq_net::Path) -> crate::Result> { + if self.waits(catalog) { + return Ok(None); + } + let mut msf = to_msf(catalog, base); + // Rebuilt from the listed tracks, so a removed track's declaration goes with it. + let mut declared = HashMap::with_capacity(msf.tracks.len()); + for track in &mut msf.tracks { + let key = (track.namespace.clone(), track.name.clone()); + let first = self.declared.remove(&key).unwrap_or_else(|| Declared::of(track)); + first.apply(track); + declared.insert(key, first); + } + self.declared = declared; + + // The MSF catalog is derived from our own types, so a serialize failure means an extension + // broke the shape; report it like any other JSON failure rather than panicking. + let json = msf.to_json().map_err(moq_json::Error::from)?; + if self.last.as_ref() == Some(&json) { + return Ok(None); + } + self.last = Some(json.clone()); + Ok(Some(json)) + } + + /// Whether an enabled rendition still has estimates to make. A disabled one promises no frames. + fn waits(&self, catalog: &Catalog) -> bool { + use std::any::TypeId; + self.unsettled.iter().any(|(ty, name)| { + if *ty == TypeId::of::() { + catalog.video.renditions.get(name).is_some_and(|config| config.enabled) + } else { + catalog.audio.renditions.get(name).is_some_and(|config| config.enabled) + } + }) + } +} + +/// The estimated fields an MSF track was first declared with. +struct Declared { + bitrate: Option, + framerate: Option, + jitter: Option, + delay: Option, +} + +impl Declared { + fn of(track: &moq_msf::Track) -> Self { + Self { + bitrate: track.bitrate, + framerate: track.framerate, + jitter: track.jitter, + delay: track.delay, + } + } + + fn apply(&self, track: &mut moq_msf::Track) { + track.bitrate = self.bitrate; + track.framerate = self.framerate; + track.jitter = self.jitter; + track.delay = self.delay; + } +} + /// The producer's `clock` under a root `clock` of `advertised`: re-based on a changed mapping while /// nothing has `anchored` the clock, and refused once something has. Clearing it is always refused. fn adopt(clock: crate::Clock, anchored: bool, advertised: Option) -> crate::Result { @@ -219,8 +305,9 @@ impl Outputs { std::borrow::Cow::Owned(catalog) } - /// Emit the catalog to hang and MSF. - fn emit(&mut self, catalog: &Catalog) -> crate::Result<()> { + /// Emit the catalog to hang and, unless `msf` holds it back, to MSF. Neither writes a catalog + /// identical to its last. + fn emit(&mut self, catalog: &Catalog, msf: &mut Msf) -> crate::Result<()> { let catalog = self.advertise(catalog); let catalog = catalog.as_ref(); // Every consumer refuses a catalog that fails this, so refuse it before it reaches one. @@ -229,16 +316,18 @@ impl Outputs { self.hang .update(moq_net::Timed::from(catalog).at(moq_net::Timestamp::now()))?; - // The MSF catalog is derived from our own types, so a serialize failure means an extension broke - // the shape; report it like any other JSON failure rather than panicking. `abort` below is - // what keeps the tracks consistent if this fails after the hang tracks were updated. - let msf = to_msf(catalog, &self.msf_track.broadcast().path) - .to_json() - .map_err(moq_json::Error::from)?; + // `abort` is what keeps the tracks consistent if this fails after the hang track was updated. + self.emit_msf(catalog, msf) + } + + /// Emit the catalog to MSF alone, unless `msf` holds it back or it matches the last one. + fn emit_msf(&mut self, catalog: &Catalog, msf: &mut Msf) -> crate::Result<()> { + let Some(msf) = msf.render(catalog, &self.msf_track.broadcast().path)? else { + return Ok(()); + }; let mut group = self.msf_track.append_group()?; group.write_frame(moq_net::Timestamp::now(), msf)?; group.finish()?; - Ok(()) } @@ -452,6 +541,7 @@ impl Producer { anchored, estimates: Throttle::default(), revisions: Default::default(), + msf: Msf::default(), })), hls: config.hls, max_age: config.max_age, @@ -760,28 +850,26 @@ impl Producer { /// Publish the buffered snapshot once, if buffering has just finished with a change staged. pub(super) fn flush_if_ready(&mut self) { - // Snapshot under the lock and emit outside it, since emitting writes to the catalog tracks. - let catalog = { - let mut state = take(&self.current); - let r = &state.reservations; - // Nothing was staged (e.g. every stream was ignored): stay unpublished so a later change - // still triggers the first emit, rather than publishing an empty catalog now. The track - // scan runs last, so releasing each of many reservations stays linear. - if r.published || r.reservers != 0 || !r.pending || !state.lists_track() { - return; - } - let r = &mut state.reservations; - r.pending = false; - r.published = true; - // A snapshot on the wire is a mapping a reader may have copied. - state.anchored = true; - state.estimates.fold(); - state.catalog.clone() - }; - if let Err(err) = self.outputs.emit(&catalog) { + let mut state = take(&self.current); + let r = &state.reservations; + // Nothing was staged (e.g. every stream was ignored): stay unpublished so a later change + // still triggers the first emit, rather than publishing an empty catalog now. The track + // scan runs last, so releasing each of many reservations stays linear. + if r.published || r.reservers != 0 || !r.pending || !state.lists_track() { + return; + } + let r = &mut state.reservations; + r.pending = false; + r.published = true; + // A snapshot on the wire is a mapping a reader may have copied. + state.anchored = true; + state.estimates.fold(); + // Emitted under the lock, as a guard's publish is, since the MSF state belongs to it. + let state = &mut *state; + if let Err(err) = self.outputs.emit(&state.catalog, &mut state.msf) { tracing::error!(%err, "failed to publish the catalog, aborting its tracks"); self.outputs.abort(&err); - take(&self.current).closed = Some(err); + state.closed = Some(err); } } @@ -927,7 +1015,8 @@ impl Producer { if state.closed.is_some() || !state.reservations.published { return; } - if let Err(err) = outputs.emit(&state.catalog) { + let state = &mut *state; + if let Err(err) = outputs.emit(&state.catalog, &mut state.msf) { tracing::error!(%err, "failed to publish an HLS declaration, aborting catalog tracks"); outputs.abort(&err); state.closed = Some(err); @@ -1121,7 +1210,8 @@ impl Guard<'_, E> { self.state.anchored = true; // Folded only once it is on the wire, so a failed emit leaves the estimate held. - self.outputs.emit(&self.state.catalog)?; + let state = &mut *self.state; + self.outputs.emit(&state.catalog, &mut state.msf)?; self.state.estimates.fold(); Ok(()) } @@ -1148,6 +1238,25 @@ impl Guard<'_, E> { self.publish() } + /// Hold the MSF catalog back until the rendition holding `name` in `C`'s section has made its + /// estimates. + pub(super) fn unsettled>(&mut self, name: &str) { + self.state.msf.unsettled.insert(owner_key::(name)); + } + + /// Stop holding the MSF catalog back for the rendition holding `name` in `C`'s section, + /// publishing it if nothing else does. + /// + /// Only MSF goes out: a hang estimate held by [`commit_estimate`](Self::commit_estimate) stays + /// held, and an initial catalog still withheld carries MSF along once its gate opens. + pub(super) fn settle>(mut self, name: &str) -> crate::Result<()> { + if !self.state.msf.unsettled.remove(&owner_key::(name)) || !self.state.reservations.published { + return Ok(()); + } + let state = &mut *self.state; + self.outputs.emit_msf(&state.catalog, &mut state.msf) + } + /// Mark the rendition holding `name` in `C`'s section as listed, so it counts toward the /// initial publish's [track requirement](State::lists_track). pub(super) fn list>(&mut self, name: &str) { @@ -1167,6 +1276,7 @@ impl Guard<'_, E> { /// name is not: nothing would ever reserve it again. pub(super) fn release>(&mut self, name: &str, present: bool) { self.state.owned.remove(&owner_key::(name)); + self.state.msf.unsettled.remove(&owner_key::(name)); if present { C::remove(&mut self.state.catalog, name); @@ -3363,4 +3473,167 @@ mod test { v.estimate(jitter(2)).unwrap(); assert_eq!(sent(&catalog), Some(0), "the gated publish opened a window"); } + + fn delay(ms: u64) -> super::super::Reading { + super::super::Reading { + estimate: super::super::Estimate::default().with_delay(std::time::Duration::from_millis(ms)), + period: Some(std::time::Duration::from_millis(10)), + framerate: false, + } + } + + /// A measured delay creeping a millisecond at a time republishes once per frame duration it + /// crosses, carrying the value measured then. Regression: every rise republished, once a second. + #[tokio::test(start_paused = true)] + async fn a_creeping_delay_republishes_once_per_frame() { + let (_broadcast, catalog, mut renditions) = estimating(&["v"]); + let v = &mut renditions[0]; + let initial = sent(&catalog).unwrap(); + + let mut published = Vec::new(); + for ms in 1..=35 { + // Past the once-a-second window, so only the frame step holds a rise back. + tokio::time::advance(std::time::Duration::from_secs(2)).await; + let before = sent(&catalog); + v.estimate(delay(ms)).unwrap(); + if sent(&catalog) != before { + published.push(wire(&catalog).video.renditions["v"].delay.unwrap().as_millis()); + } + } + assert_eq!(published, [10, 20, 30]); + assert_eq!(sent(&catalog), Some(initial + 3)); + assert_eq!( + catalog.snapshot().video.renditions["v"].delay, + Some(std::time::Duration::from_millis(30)), + "a rise short of a frame is not even staged" + ); + } + + /// A known framerate sets the step, whatever shorter advance the timestamps show. + #[tokio::test(start_paused = true)] + async fn a_framerate_sets_the_delay_step() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); + let reserved = catalog.reserve(); + let mut v = reserved.init::("v").unwrap(); + let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); + config.framerate = Some(25.0); + v.set(config).unwrap(); + drop(reserved); + + v.estimate(delay(39)).unwrap(); + assert_eq!(wire(&catalog).video.renditions["v"].delay, None); + v.estimate(delay(40)).unwrap(); + assert_eq!( + wire(&catalog).video.renditions["v"].delay, + Some(std::time::Duration::from_millis(40)) + ); + } + + /// How many MSF catalogs have gone out on the wire. + fn msf_sent(catalog: &Producer) -> Option { + catalog.outputs.msf_track.largest().map(|position| position.group) + } + + /// The last MSF catalog on the wire. + fn msf_wire(catalog: &Producer) -> moq_msf::Catalog<()> { + let state = take(&catalog.current); + moq_msf::Catalog::from_str(state.msf.last.as_deref().expect("a published MSF catalog")).unwrap() + } + + /// A change MSF doesn't carry republishes hang but not MSF. + #[test] + fn identical_msf_content_publishes_once() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let mut catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); + catalog + .mutate(|catalog| { + catalog + .video + .renditions + .insert("v".to_string(), VideoConfig::new(hang::catalog::VideoCodec::VP8)); + }) + .unwrap(); + assert_eq!(msf_sent(&catalog), Some(0)); + let hang = sent(&catalog); + + catalog + .mutate(|catalog| { + catalog.video.display = Some(hang::catalog::Display { + width: 1280, + height: 720, + }) + }) + .unwrap(); + assert_ne!(sent(&catalog), hang); + assert_eq!(msf_sent(&catalog), Some(0)); + } + + /// An unhinted track holds the MSF catalog back until its first groups close over a second of + /// media, so even 20 ms audio groups declare a bitrate, which later estimates leave alone. + #[tokio::test(start_paused = true)] + async fn msf_waits_for_estimates_and_freezes_them() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); + let reserved = catalog.reserve(); + let mut audio = reserved.init::("a").unwrap(); + audio + .set(AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2)) + .unwrap(); + drop(reserved); + assert_eq!(sent(&catalog), Some(0), "hang publishes at once"); + assert_eq!(msf_sent(&catalog), None); + + let mut estimator = super::super::Estimator::new(); + let frame = |estimator: &mut super::super::Estimator, i: u64, bytes: usize| { + let ts = moq_net::Timestamp::from_micros(i * 20_000).unwrap(); + estimator.cut(Some(ts)); + estimator.write(ts, bytes); + }; + for i in 0..50 { + frame(&mut estimator, i, 100); + audio.estimate(estimator.reading()).unwrap(); + assert_eq!(msf_sent(&catalog), None, "frame {i} is inside the first second"); + } + frame(&mut estimator, 50, 100); + audio.estimate(estimator.reading()).unwrap(); + assert_eq!(msf_sent(&catalog), Some(0)); + let declared = msf_wire(&catalog).tracks[0].bitrate; + assert_eq!(declared, Some(40_000)); + + // A louder second raises the hang bitrate; MSF keeps what it declared. + for i in 51..200 { + frame(&mut estimator, i, 400); + audio.estimate(estimator.reading()).unwrap(); + } + assert!(wire(&catalog).audio.renditions["a"].bitrate > declared); + assert_eq!(msf_sent(&catalog), Some(0)); + assert_eq!(msf_wire(&catalog).tracks[0].bitrate, declared); + } + + /// A track whose every estimated field is hinted has nothing to wait for. + #[test] + fn a_fully_hinted_track_publishes_msf_at_once() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); + let reserved = catalog.reserve(); + let mut video = reserved.init::("v").unwrap(); + let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); + config.bitrate = Some(1_000_000); + config.framerate = Some(30.0); + config.jitter = Some(std::time::Duration::from_millis(1)); + config.delay = Some(std::time::Duration::from_millis(1)); + video.set(config).unwrap(); + drop(reserved); + assert_eq!(msf_sent(&catalog), Some(0)); + } + + /// A track that ends before its estimates are made stops holding the MSF catalog back. + #[test] + fn a_settled_track_releases_msf() { + let (_broadcast, catalog, mut renditions) = estimating(&["v"]); + assert_eq!(msf_sent(&catalog), None); + renditions[0].settle().unwrap(); + assert_eq!(msf_sent(&catalog), Some(0)); + } } diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index ff8aca1ff8b..ed97ef50718 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -367,6 +367,43 @@ impl RenditionConfig for hang::catalog::TextConfig { } } +/// How far short of a whole frame duration a value still counts as reaching it, as a fraction of the +/// period: a reorder of exactly one frame must not miss its step to rounding in the frame duration. +const STEP_SLACK: u32 = 100; + +/// `measured` if it reaches a whole `period` beyond the one `published` sits in, else `published`. +fn quantize(published: Option, measured: Option, period: Duration) -> Option { + let step = + |value: Option| value.map_or(0, |value| (value + period / STEP_SLACK).as_nanos() / period.as_nanos()); + if step(measured) > step(published) { + measured + } else { + published + } +} + +/// Whether config `C` is listed as an MSF track whose estimates the MSF catalog waits for, and if so +/// whether it carries a framerate. Text detects nothing and data tracks aren't listed in MSF. +fn msf_framerate() -> Option { + use std::any::TypeId; + let ty = TypeId::of::(); + if ty == TypeId::of::() { + Some(true) + } else if ty == TypeId::of::() { + Some(false) + } else { + None + } +} + +/// Whether `supplied` hints every field an MSF track's estimate would otherwise wait on. +fn hinted(supplied: &Estimate, framerate: bool) -> bool { + supplied.jitter.is_some() + && supplied.bitrate.is_some() + && supplied.delay.is_some() + && (!framerate || supplied.framerate.is_some()) +} + /// Refuse an edit that redefines rendition `name`'s track identity in place: a publisher whose /// decoder identity changed publishes under a new name instead. fn check_identity(previous: &Catalog, next: &Catalog, name: &str) -> crate::Result<()> { @@ -556,6 +593,8 @@ pub(crate) struct Rendition> { detected: Estimate, /// The last estimate written to the config, so an unchanged one doesn't republish the catalog. published: Option, + /// Whether the estimates the MSF catalog freezes have been made, so it no longer waits on them. + settled: bool, _config: PhantomData C>, } @@ -604,6 +643,7 @@ impl> Rendition { supplied: Estimate::default(), detected: Estimate::default(), published: None, + settled: false, _config: PhantomData, }) } @@ -650,6 +690,11 @@ impl> Rendition { self.present = true; *guard = next; guard.list::(&self.name); + // Every estimated field hinted leaves nothing to wait for. + self.settled |= msf_framerate::().is_none_or(|framerate| hinted(&supplied, framerate)); + if !self.settled { + guard.unsettled::(&self.name); + } guard.commit()?; } self.supplied = supplied; @@ -679,15 +724,21 @@ impl> Rendition { /// Publish a measured [`Estimate`], filling only the fields the config didn't supply at /// [`set`](Self::set). /// - /// Mint the estimate from an [`Estimator`](super::Estimator), usually the one owned by a + /// Mint the reading from an [`Estimator`](super::Estimator), usually the one owned by a /// [`container::Producer`](crate::container::Producer). Call it after every write: an estimate - /// that resolves to what the catalog already carries doesn't republish it, and a `jitter` or - /// `delay` rise republishes at most once a second (see - /// [`Guard::commit_estimate`](super::Guard::commit_estimate)), going out on a later call. + /// that resolves to what the catalog already carries doesn't republish it. A `jitter` or `delay` + /// rise republishes only once it reaches the next whole frame duration above the published value, + /// and then at most once a second (see [`Guard::commit_estimate`](super::Guard::commit_estimate)), + /// going out on a later call. /// /// Calling this before [`set`](Self::set) is not wasted: the measurement is remembered and seeds /// the config once it lands. - pub(crate) fn estimate(&mut self, estimate: Estimate) -> crate::Result<()> { + pub(crate) fn estimate(&mut self, reading: impl Into) -> crate::Result<()> { + let super::Reading { + estimate, + period, + framerate: rated, + } = reading.into(); if !C::detects() { return Ok(()); } @@ -695,6 +746,31 @@ impl> Rendition { self.detected = estimate; return Ok(()); } + // The bitrate's first window closes on the first group boundary after a window of media, and + // a framerate to measure takes its own window after that. + let framerate = self.supplied.framerate.is_none() && msf_framerate::() == Some(true); + let settles = !self.settled && estimate.bitrate.is_some() && (!framerate || rated); + self.publish(estimate, period)?; + if settles { + self.settle()?; + } + Ok(()) + } + + /// Stop holding the MSF catalog back for this rendition's estimates: they are made, or the track + /// has ended and they never will be. + pub(crate) fn settle(&mut self) -> crate::Result<()> { + if std::mem::replace(&mut self.settled, true) || !self.present { + return Ok(()); + } + // A closed catalog has nothing left to publish. + if let Ok(guard) = self.catalog.modify() { + guard.settle::(&self.name)?; + } + Ok(()) + } + + fn publish(&mut self, estimate: Estimate, period: Option) -> crate::Result<()> { let mut resolved = Self::resolved(&self.supplied, &estimate); // A measurement never lowers the published jitter or delay, including one raised through // `modify`. @@ -702,6 +778,18 @@ impl> Rendition { let published = config.estimate(); resolved.jitter = resolved.jitter.max(published.jitter); resolved.delay = resolved.delay.max(published.delay); + // Measurement creeps up as the encoder's latency does, a step per frame at worst; only a whole + // frame's rise is worth a republish. Until a period is known the throttle alone paces it. + let period = resolved + .framerate + .filter(|framerate| *framerate > 0.0) + .and_then(|framerate| Duration::try_from_secs_f64(framerate.recip()).ok()) + .or(period) + .filter(|period| !period.is_zero()); + if let Some(period) = period { + resolved.jitter = quantize(published.jitter, resolved.jitter, period); + resolved.delay = quantize(published.delay, resolved.delay, period); + } self.detected = estimate; if self.published.as_ref() == Some(&resolved) { // Still an observation: it releases a rise held back until its window ended. A closed @@ -1092,7 +1180,7 @@ mod tests { let t = ts(i * 40_000); estimator.cut(Some(t)); estimator.write(t, 100_000); - rendition.estimate(estimator.estimate()).unwrap(); + rendition.estimate(estimator.reading()).unwrap(); } } diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index bb676a23ba0..08c99758cdc 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -99,10 +99,17 @@ enum Rendition { } impl Rendition { - fn estimate(&mut self, estimate: crate::catalog::Estimate) -> crate::Result<()> { + fn estimate(&mut self, reading: crate::catalog::Reading) -> crate::Result<()> { match self { - Self::Video(rendition) => rendition.estimate(estimate), - Self::Audio(rendition) => rendition.estimate(estimate), + Self::Video(rendition) => rendition.estimate(reading), + Self::Audio(rendition) => rendition.estimate(reading), + } + } + + fn settle(&mut self) -> crate::Result<()> { + match self { + Self::Video(rendition) => rendition.settle(), + Self::Audio(rendition) => rendition.settle(), } } } @@ -179,9 +186,9 @@ struct Passthrough { impl Passthrough { fn publish_estimate(&mut self) -> crate::Result<()> { - let estimate = self.estimator.estimate(); - self.claim.update(&self.track.demand(), estimate.bitrate); - self.rendition.estimate(estimate) + let reading = self.estimator.reading(); + self.claim.update(&self.track.demand(), reading.estimate.bitrate); + self.rendition.estimate(reading) } /// Close the open group, so the next fragment opens one at `pending_sequence` or the next. @@ -198,6 +205,8 @@ impl Passthrough { fn finish(&mut self) -> crate::Result<()> { self.close()?; self.track.finish()?; + // Its estimates are as made as they will be, so the MSF catalog stops waiting on them. + self.rendition.settle()?; // The importer may be kept around, but its media stops holding the other tracks back. self.estimator.discontinuity(); // Dropping the recorder stops the open HLS segment, so the track is advertised. diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 5a3c6c1835a..4309e8363eb 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -1381,6 +1381,9 @@ async fn test_msf_catalog_roundtrip() { let buf = bytes::BytesMut::from(&data[..]); // Trailing fragments may error out (e.g. partial mdat); ignore. let _ = fmp4.decode(&buf); + // The clip is shorter than a second, so its estimates are never made: the MSF catalog waits for + // the tracks to end. + fmp4.finish().unwrap(); let track = consumer .track(moq_msf::DEFAULT_NAME) diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index afc6802b9b8..2b4147df7f2 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -107,7 +107,8 @@ trait Rendition: Send { fn set(&mut self, config: R) -> crate::Result<()>; fn config(&self) -> crate::Result; fn replace(&mut self, config: R) -> crate::Result<()>; - fn estimate(&mut self, estimate: crate::catalog::Estimate) -> crate::Result<()>; + fn estimate(&mut self, reading: crate::catalog::Reading) -> crate::Result<()>; + fn settle(&mut self) -> crate::Result<()>; fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()>; fn fresh( &self, @@ -156,8 +157,12 @@ where self.replace(config) } - fn estimate(&mut self, estimate: crate::catalog::Estimate) -> crate::Result<()> { - self.estimate(estimate) + fn estimate(&mut self, reading: crate::catalog::Reading) -> crate::Result<()> { + self.estimate(reading) + } + + fn settle(&mut self) -> crate::Result<()> { + self.settle() } fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()> { @@ -270,7 +275,7 @@ where fn publish_estimate(&mut self) -> crate::Result<()> { if let Some(rendition) = self.rendition.as_mut() { - rendition.estimate(self.estimator.estimate())?; + rendition.estimate(self.estimator.reading())?; } Ok(()) } @@ -804,6 +809,10 @@ where self.cut(None)?; self.inner.finish()?; self.publish_estimate()?; + // Its estimates are as made as they will be, so the MSF catalog stops waiting on them. + if let Some(rendition) = self.rendition.as_mut() { + rendition.settle()?; + } // A finished track may be kept around, but its media stops holding the other tracks back. self.estimator.discontinuity(); // Dropping the recorder stops the open HLS segment, so a track that ends before its catalog From 51aca44ea3f24e8e0b6df19fcae2d7200afd2b70 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 17:20:18 -0700 Subject: [PATCH 3/5] fix(mux): settle unhinted video on its bitrate window, as planned MSF publishes once every unhinted track has closed its first group and carried a second of media, with no extra wait for the video framerate window. Unhinted video now declares no framerate in MSF. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/catalog/estimate.rs | 14 +------------- rs/moq-mux/src/catalog/producer.rs | 28 +++++++++++++++++++++++++++- rs/moq-mux/src/catalog/tracks.rs | 13 ++++--------- 3 files changed, 32 insertions(+), 23 deletions(-) diff --git a/rs/moq-mux/src/catalog/estimate.rs b/rs/moq-mux/src/catalog/estimate.rs index c4da7393249..bec51470bf3 100644 --- a/rs/moq-mux/src/catalog/estimate.rs +++ b/rs/moq-mux/src/catalog/estimate.rs @@ -195,7 +195,6 @@ impl Estimator { Reading { estimate: self.estimate(), period: self.framerate.shortest, - framerate: self.framerate.measured, } } @@ -220,18 +219,11 @@ pub(crate) struct Reading { /// The shortest advance between frames, which a published `jitter` or `delay` rises in whole /// steps of when no framerate says the frame duration. pub period: Option, - /// Whether the framerate's first window has closed. It opens on the first group boundary and - /// closes a [`RATE_WINDOW`] later. - pub framerate: bool, } impl From for Reading { fn from(estimate: Estimate) -> Self { - Self { - estimate, - period: None, - framerate: false, - } + Self { estimate, period: None } } } @@ -338,8 +330,6 @@ struct Framerate { /// Frames written since the window opened. frames: u64, max: Option, - /// Whether a window has closed, whether or not its rate snapped. - measured: bool, /// The shortest advance of the highest timestamp: a frame's duration, or a reordered group's /// span of them. shortest: Option, @@ -380,7 +370,6 @@ impl Framerate { { self.max = Some(framerate); } - self.measured = true; self.open = Some(high); self.frames = 0; } @@ -389,7 +378,6 @@ impl Framerate { fn discontinuity(&mut self) { *self = Self { max: self.max, - measured: self.measured, shortest: self.shortest, ..Self::default() }; diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 08f7b67df48..39e78a557af 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -3478,7 +3478,6 @@ mod test { super::super::Reading { estimate: super::super::Estimate::default().with_delay(std::time::Duration::from_millis(ms)), period: Some(std::time::Duration::from_millis(10)), - framerate: false, } } @@ -3611,6 +3610,33 @@ mod test { assert_eq!(msf_wire(&catalog).tracks[0].bitrate, declared); } + /// Unhinted video settles with its bitrate, before its framerate window has closed, so MSF + /// declares the framerate absent and keeps it absent once hang measures one. + #[tokio::test(start_paused = true)] + async fn unhinted_video_declares_no_framerate() { + let (_broadcast, catalog, mut renditions) = estimating(&["v"]); + let v = &mut renditions[0]; + + // 25 fps in 2 s groups. + let mut estimator = super::super::Estimator::new(); + for i in 0..=100u64 { + let ts = moq_net::Timestamp::from_micros(i * 40_000).unwrap(); + if i % 50 == 0 { + estimator.cut(Some(ts)); + } + estimator.write(ts, 1_000); + v.estimate(estimator.reading()).unwrap(); + if i < 50 { + assert_eq!(msf_sent(&catalog), None, "frame {i} is inside the first group"); + } + } + assert_eq!(wire(&catalog).video.renditions["v"].framerate, Some(25.0)); + assert_eq!(msf_sent(&catalog), Some(0)); + let track = &msf_wire(&catalog).tracks[0]; + assert!(track.bitrate.is_some()); + assert_eq!(track.framerate, None); + } + /// A track whose every estimated field is hinted has nothing to wait for. #[test] fn a_fully_hinted_track_publishes_msf_at_once() { diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index ed97ef50718..d4f0da9c313 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -734,11 +734,7 @@ impl> Rendition { /// Calling this before [`set`](Self::set) is not wasted: the measurement is remembered and seeds /// the config once it lands. pub(crate) fn estimate(&mut self, reading: impl Into) -> crate::Result<()> { - let super::Reading { - estimate, - period, - framerate: rated, - } = reading.into(); + let super::Reading { estimate, period } = reading.into(); if !C::detects() { return Ok(()); } @@ -746,10 +742,9 @@ impl> Rendition { self.detected = estimate; return Ok(()); } - // The bitrate's first window closes on the first group boundary after a window of media, and - // a framerate to measure takes its own window after that. - let framerate = self.supplied.framerate.is_none() && msf_framerate::() == Some(true); - let settles = !self.settled && estimate.bitrate.is_some() && (!framerate || rated); + // The bitrate's first window closes on the first group boundary after a window of media. An + // unhinted framerate takes its own window after that, so MSF declares it absent. + let settles = !self.settled && estimate.bitrate.is_some(); self.publish(estimate, period)?; if settles { self.settle()?; From 945ceaedaba848c92d96c2f54a32b29e12c5d3cc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 17:27:59 -0700 Subject: [PATCH 4/5] fix(mux): a fully hinted replacement config releases the MSF hold Replacing an unhinted config with one hinting every estimated field marked the rendition settled but left the catalog waiting on it, withholding MSF for as long as the track stayed listed. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/catalog/producer.rs | 29 +++++++++++++++++++++++++---- rs/moq-mux/src/catalog/tracks.rs | 5 ++--- 2 files changed, 27 insertions(+), 7 deletions(-) diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 39e78a557af..ca11eb5de06 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -1238,10 +1238,15 @@ impl Guard<'_, E> { self.publish() } - /// Hold the MSF catalog back until the rendition holding `name` in `C`'s section has made its - /// estimates. - pub(super) fn unsettled>(&mut self, name: &str) { - self.state.msf.unsettled.insert(owner_key::(name)); + /// Whether the MSF catalog waits for the rendition holding `name` in `C`'s section to make its + /// estimates. Released, it goes out with the next publish. + pub(super) fn unsettled>(&mut self, name: &str, unsettled: bool) { + let key = owner_key::(name); + if unsettled { + self.state.msf.unsettled.insert(key); + } else { + self.state.msf.unsettled.remove(&key); + } } /// Stop holding the MSF catalog back for the rendition holding `name` in `C`'s section, @@ -3654,6 +3659,22 @@ mod test { assert_eq!(msf_sent(&catalog), Some(0)); } + /// Replacing an unhinted config with one hinting every estimated field releases the MSF hold. + /// Regression: the rendition counted itself settled while the catalog still waited on it. + #[test] + fn a_fully_hinted_replacement_releases_msf() { + let (_broadcast, catalog, mut renditions) = estimating(&["v"]); + assert_eq!(msf_sent(&catalog), None); + let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); + config.bitrate = Some(1_000_000); + config.framerate = Some(30.0); + config.jitter = Some(std::time::Duration::from_millis(1)); + config.delay = Some(std::time::Duration::from_millis(1)); + renditions[0].set(config).unwrap(); + assert_eq!(msf_sent(&catalog), Some(0)); + assert_eq!(msf_wire(&catalog).tracks[0].framerate, Some(30.0)); + } + /// A track that ends before its estimates are made stops holding the MSF catalog back. #[test] fn a_settled_track_releases_msf() { diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index d4f0da9c313..1fbb96ef963 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -692,9 +692,8 @@ impl> Rendition { guard.list::(&self.name); // Every estimated field hinted leaves nothing to wait for. self.settled |= msf_framerate::().is_none_or(|framerate| hinted(&supplied, framerate)); - if !self.settled { - guard.unsettled::(&self.name); - } + // A replaced config that hints everything releases the hold its predecessor took. + guard.unsettled::(&self.name, !self.settled); guard.commit()?; } self.supplied = supplied; From fde6e88ce9ab526ad3388e3df29ceba008cb8c30 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 17:57:28 -0700 Subject: [PATCH 5/5] fix(mux): MSF declares only hinted estimates MSF held its catalog back until every unhinted track was measured, which deadlocks demand-driven publishers: they produce no media until a viewer subscribes, and an MSF viewer finds their tracks only through MSF. MSF now declares only the estimates a publisher hinted and publishes at once; hang keeps its measured estimates. Co-Authored-By: Claude Opus 5.5 --- rs/moq-c/src/api.rs | 3 +- rs/moq-ffi/src/media.rs | 1 + rs/moq-mux/src/catalog/producer.rs | 232 +++++++------------ rs/moq-mux/src/catalog/tracks.rs | 78 ++----- rs/moq-mux/src/container/fmp4/import.rs | 9 - rs/moq-mux/src/container/fmp4/import_test.rs | 3 - rs/moq-mux/src/container/producer.rs | 9 - 7 files changed, 110 insertions(+), 225 deletions(-) diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 81a8dd8190b..628fb5aad0d 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -206,7 +206,8 @@ pub struct moq_video_init { /// /// Zero the struct and set only the `has_*` flags you want. Hints fill gaps the /// bitstream leaves (especially bitrate); a value the stream detects later wins -/// for dimensions. +/// for dimensions. The MSF catalog declares a bitrate or frame rate only when +/// it is hinted here. #[repr(C)] #[derive(Clone, Copy, Default)] pub struct moq_video_hint { diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index c3b5397aba8..8d8bc3e805a 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -228,6 +228,7 @@ pub struct MoqDatagram { /// Caller-provided video catalog fields for [`MoqVideoInit`]. /// /// Every field is optional and fills only a gap the stream leaves; a value the stream detects wins. +/// The MSF catalog declares a bitrate or frame rate only when it is hinted here. /// Publishing the catalog before the first keyframe needs at least the codec, which comes from the /// [`MoqVideoInit`] format. Audio has no equivalent: it resolves entirely from its init bytes. #[derive(Clone, Default, uniffi::Record)] diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index ca11eb5de06..96439aa5fe1 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -54,8 +54,8 @@ struct State { /// reuses a name. revisions: std::collections::HashMap, - /// What the MSF catalog has declared, and what it still waits for. - msf: Msf, + /// What the MSF catalog declares. + msf: Msf, } impl State { @@ -114,28 +114,49 @@ impl Throttle { /// The MSF catalog's publish state. /// -/// MSF-01 §5.3 fixes a declared track's attributes, so the MSF catalog waits for each track's -/// estimates to be made, declares them once, and then changes only when the track list or a track's -/// identity does. The hang catalog keeps updating its estimates in place. -#[derive(Default)] -struct Msf { - /// The listed video and audio renditions whose estimates are still being made, keyed like - /// [`State::owned`]. Any enabled one holds the MSF catalog back. - unsettled: std::collections::BTreeSet<(std::any::TypeId, String)>, +/// MSF declares only the estimates a publisher hinted, never measured ones: a demand-driven +/// publisher measures nothing until a viewer subscribes, and an MSF viewer finds its tracks only +/// through this catalog. MSF-01 §5.3 fixes a declared track's attributes, so each track keeps the +/// estimates it was first declared with, and the MSF catalog changes only when the track list or a +/// track's identity does. The hang catalog keeps updating its estimates in place. +struct Msf { + /// The hinted estimates of each listed rendition that measures its own, keyed like + /// [`State::owned`]. MSF declares these in place of what the catalog carries. + hints: HashMap<(std::any::TypeId, String), Hint>, /// The estimates each MSF track was first declared with, by namespace and name. declared: HashMap<(Option, String), Declared>, /// The last MSF catalog written, so an identical one isn't written again. last: Option, } -impl Msf { - /// The MSF catalog to write for `catalog`, or `None` while a track's estimates are still being - /// made or when it matches the last one written. - fn render(&mut self, catalog: &Catalog, base: &moq_net::Path) -> crate::Result> { - if self.waits(catalog) { - return Ok(None); +impl Default for Msf { + fn default() -> Self { + Self { + hints: HashMap::new(), + declared: HashMap::new(), + last: None, } - let mut msf = to_msf(catalog, base); + } +} + +/// A rendition's hinted estimates, and how to write them into its config. +struct Hint { + estimate: super::Estimate, + apply: fn(&mut Catalog, &str, super::Estimate), +} + +impl Msf { + /// The MSF catalog to write for `catalog`, or `None` when it matches the last one written. + fn render(&mut self, catalog: &Catalog, base: &moq_net::Path) -> crate::Result> { + let mut msf = if self.hints.is_empty() { + to_msf(catalog, base) + } else { + let mut hinted = catalog.clone(); + for ((_, name), hint) in &self.hints { + (hint.apply)(&mut hinted, name, hint.estimate.clone()); + } + to_msf(&hinted, base) + }; // Rebuilt from the listed tracks, so a removed track's declaration goes with it. let mut declared = HashMap::with_capacity(msf.tracks.len()); for track in &mut msf.tracks { @@ -155,17 +176,16 @@ impl Msf { self.last = Some(json.clone()); Ok(Some(json)) } +} - /// Whether an enabled rendition still has estimates to make. A disabled one promises no frames. - fn waits(&self, catalog: &Catalog) -> bool { - use std::any::TypeId; - self.unsettled.iter().any(|(ty, name)| { - if *ty == TypeId::of::() { - catalog.video.renditions.get(name).is_some_and(|config| config.enabled) - } else { - catalog.audio.renditions.get(name).is_some_and(|config| config.enabled) - } - }) +/// Write `estimate` over the estimates of the `C` config under `name`. +fn apply_hint>( + catalog: &mut Catalog, + name: &str, + estimate: super::Estimate, +) { + if let Some(config) = C::get_mut(catalog, name) { + config.set_estimate(estimate); } } @@ -305,9 +325,8 @@ impl Outputs { std::borrow::Cow::Owned(catalog) } - /// Emit the catalog to hang and, unless `msf` holds it back, to MSF. Neither writes a catalog - /// identical to its last. - fn emit(&mut self, catalog: &Catalog, msf: &mut Msf) -> crate::Result<()> { + /// Emit the catalog to hang and MSF. Neither writes a catalog identical to its last. + fn emit(&mut self, catalog: &Catalog, msf: &mut Msf) -> crate::Result<()> { let catalog = self.advertise(catalog); let catalog = catalog.as_ref(); // Every consumer refuses a catalog that fails this, so refuse it before it reaches one. @@ -316,18 +335,15 @@ impl Outputs { self.hang .update(moq_net::Timed::from(catalog).at(moq_net::Timestamp::now()))?; - // `abort` is what keeps the tracks consistent if this fails after the hang track was updated. - self.emit_msf(catalog, msf) - } - - /// Emit the catalog to MSF alone, unless `msf` holds it back or it matches the last one. - fn emit_msf(&mut self, catalog: &Catalog, msf: &mut Msf) -> crate::Result<()> { + // `abort` below is what keeps the tracks consistent if this fails after the hang track was + // updated. let Some(msf) = msf.render(catalog, &self.msf_track.broadcast().path)? else { return Ok(()); }; let mut group = self.msf_track.append_group()?; group.write_frame(moq_net::Timestamp::now(), msf)?; group.finish()?; + Ok(()) } @@ -1238,28 +1254,14 @@ impl Guard<'_, E> { self.publish() } - /// Whether the MSF catalog waits for the rendition holding `name` in `C`'s section to make its - /// estimates. Released, it goes out with the next publish. - pub(super) fn unsettled>(&mut self, name: &str, unsettled: bool) { - let key = owner_key::(name); - if unsettled { - self.state.msf.unsettled.insert(key); - } else { - self.state.msf.unsettled.remove(&key); - } - } - - /// Stop holding the MSF catalog back for the rendition holding `name` in `C`'s section, - /// publishing it if nothing else does. - /// - /// Only MSF goes out: a hang estimate held by [`commit_estimate`](Self::commit_estimate) stays - /// held, and an initial catalog still withheld carries MSF along once its gate opens. - pub(super) fn settle>(mut self, name: &str) -> crate::Result<()> { - if !self.state.msf.unsettled.remove(&owner_key::(name)) || !self.state.reservations.published { - return Ok(()); - } - let state = &mut *self.state; - self.outputs.emit_msf(&state.catalog, &mut state.msf) + /// Declare `estimate` in MSF for the rendition holding `name` in `C`'s section, in place of + /// whatever it measures. It goes out with the next publish. + pub(super) fn hint>(&mut self, name: &str, estimate: super::Estimate) { + let hint = Hint { + estimate, + apply: apply_hint::, + }; + self.state.msf.hints.insert(owner_key::(name), hint); } /// Mark the rendition holding `name` in `C`'s section as listed, so it counts toward the @@ -1281,7 +1283,7 @@ impl Guard<'_, E> { /// name is not: nothing would ever reserve it again. pub(super) fn release>(&mut self, name: &str, present: bool) { self.state.owned.remove(&owner_key::(name)); - self.state.msf.unsettled.remove(&owner_key::(name)); + self.state.msf.hints.remove(&owner_key::(name)); if present { C::remove(&mut self.state.catalog, name); @@ -3573,114 +3575,58 @@ mod test { assert_eq!(msf_sent(&catalog), Some(0)); } - /// An unhinted track holds the MSF catalog back until its first groups close over a second of - /// media, so even 20 ms audio groups declare a bitrate, which later estimates leave alone. + /// A demand-driven track measures nothing until a viewer subscribes, which an MSF viewer can do + /// only once MSF lists it. So an unhinted track declares MSF at once without estimates, and they + /// stay absent once hang measures them. Regression: MSF waited for the estimates. #[tokio::test(start_paused = true)] - async fn msf_waits_for_estimates_and_freezes_them() { - let mut broadcast = moq_net::broadcast::Info::new().produce(); - let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); - let reserved = catalog.reserve(); - let mut audio = reserved.init::("a").unwrap(); - audio - .set(AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2)) - .unwrap(); - drop(reserved); - assert_eq!(sent(&catalog), Some(0), "hang publishes at once"); - assert_eq!(msf_sent(&catalog), None); - - let mut estimator = super::super::Estimator::new(); - let frame = |estimator: &mut super::super::Estimator, i: u64, bytes: usize| { - let ts = moq_net::Timestamp::from_micros(i * 20_000).unwrap(); - estimator.cut(Some(ts)); - estimator.write(ts, bytes); - }; - for i in 0..50 { - frame(&mut estimator, i, 100); - audio.estimate(estimator.reading()).unwrap(); - assert_eq!(msf_sent(&catalog), None, "frame {i} is inside the first second"); - } - frame(&mut estimator, 50, 100); - audio.estimate(estimator.reading()).unwrap(); - assert_eq!(msf_sent(&catalog), Some(0)); - let declared = msf_wire(&catalog).tracks[0].bitrate; - assert_eq!(declared, Some(40_000)); - - // A louder second raises the hang bitrate; MSF keeps what it declared. - for i in 51..200 { - frame(&mut estimator, i, 400); - audio.estimate(estimator.reading()).unwrap(); - } - assert!(wire(&catalog).audio.renditions["a"].bitrate > declared); - assert_eq!(msf_sent(&catalog), Some(0)); - assert_eq!(msf_wire(&catalog).tracks[0].bitrate, declared); - } - - /// Unhinted video settles with its bitrate, before its framerate window has closed, so MSF - /// declares the framerate absent and keeps it absent once hang measures one. - #[tokio::test(start_paused = true)] - async fn unhinted_video_declares_no_framerate() { + async fn an_unhinted_track_declares_msf_at_once() { let (_broadcast, catalog, mut renditions) = estimating(&["v"]); let v = &mut renditions[0]; + assert_eq!(msf_sent(&catalog), Some(0), "no media has been written"); + let track = &msf_wire(&catalog).tracks[0]; + assert_eq!((track.bitrate, track.framerate), (None, None)); + assert_eq!((track.jitter, track.delay), (None, None)); // 25 fps in 2 s groups. let mut estimator = super::super::Estimator::new(); - for i in 0..=100u64 { + for i in 0..=150u64 { let ts = moq_net::Timestamp::from_micros(i * 40_000).unwrap(); if i % 50 == 0 { estimator.cut(Some(ts)); } estimator.write(ts, 1_000); v.estimate(estimator.reading()).unwrap(); - if i < 50 { - assert_eq!(msf_sent(&catalog), None, "frame {i} is inside the first group"); - } } - assert_eq!(wire(&catalog).video.renditions["v"].framerate, Some(25.0)); - assert_eq!(msf_sent(&catalog), Some(0)); - let track = &msf_wire(&catalog).tracks[0]; - assert!(track.bitrate.is_some()); - assert_eq!(track.framerate, None); + let hang = &wire(&catalog).video.renditions["v"]; + assert!(hang.bitrate.is_some()); + assert_eq!(hang.framerate, Some(25.0)); + assert_eq!(msf_sent(&catalog), Some(0), "measurements leave MSF alone"); } - /// A track whose every estimated field is hinted has nothing to wait for. - #[test] - fn a_fully_hinted_track_publishes_msf_at_once() { + /// MSF declares the estimates a track hinted, and keeps those it first declared when a hint + /// changes. A field left unhinted stays absent whatever hang measures. + #[tokio::test(start_paused = true)] + async fn a_hinted_track_carries_its_hints() { let mut broadcast = moq_net::broadcast::Info::new().produce(); let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); let reserved = catalog.reserve(); - let mut video = reserved.init::("v").unwrap(); + let mut v = reserved.init::("v").unwrap(); let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); config.bitrate = Some(1_000_000); config.framerate = Some(30.0); - config.jitter = Some(std::time::Duration::from_millis(1)); - config.delay = Some(std::time::Duration::from_millis(1)); - video.set(config).unwrap(); + v.set(config).unwrap(); drop(reserved); assert_eq!(msf_sent(&catalog), Some(0)); - } - - /// Replacing an unhinted config with one hinting every estimated field releases the MSF hold. - /// Regression: the rendition counted itself settled while the catalog still waited on it. - #[test] - fn a_fully_hinted_replacement_releases_msf() { - let (_broadcast, catalog, mut renditions) = estimating(&["v"]); - assert_eq!(msf_sent(&catalog), None); - let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); - config.bitrate = Some(1_000_000); - config.framerate = Some(30.0); - config.jitter = Some(std::time::Duration::from_millis(1)); - config.delay = Some(std::time::Duration::from_millis(1)); - renditions[0].set(config).unwrap(); - assert_eq!(msf_sent(&catalog), Some(0)); - assert_eq!(msf_wire(&catalog).tracks[0].framerate, Some(30.0)); - } + let track = &msf_wire(&catalog).tracks[0]; + assert_eq!((track.bitrate, track.framerate), (Some(1_000_000), Some(30.0))); - /// A track that ends before its estimates are made stops holding the MSF catalog back. - #[test] - fn a_settled_track_releases_msf() { - let (_broadcast, catalog, mut renditions) = estimating(&["v"]); - assert_eq!(msf_sent(&catalog), None); - renditions[0].settle().unwrap(); + v.estimate(delay(40)).unwrap(); + v.supply_bitrate(2_000_000).unwrap(); + let hang = &wire(&catalog).video.renditions["v"]; + assert_eq!(hang.delay, Some(std::time::Duration::from_millis(40))); + assert_eq!(hang.bitrate, Some(2_000_000)); assert_eq!(msf_sent(&catalog), Some(0)); + let track = &msf_wire(&catalog).tracks[0]; + assert_eq!((track.bitrate, track.delay), (Some(1_000_000), None)); } } diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index 1fbb96ef963..95b1a8cf1df 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -171,7 +171,8 @@ impl RenditionConfig for hang::catalog::BinaryConfig { /// dimensions from an SPS, ...) always wins, so a detected change just updates the catalog. A /// [`codec`](Self::codec) alone is enough to publish the catalog before the first keyframe (the /// decoder config then arrives in band). Use it for fields the stream can't reveal (bitrate) or to -/// skip that wait. +/// skip that wait. The MSF catalog declares only hinted estimates (`bitrate`, `framerate`, `jitter`, +/// `delay`), never measured ones, so an MSF track carries only those set here. /// /// Video importers take one because they build the [`VideoConfig`](hang::catalog::VideoConfig) /// themselves, out of the bitstream, so the caller never holds the config to set these on. An audio @@ -382,28 +383,6 @@ fn quantize(published: Option, measured: Option, period: Dur } } -/// Whether config `C` is listed as an MSF track whose estimates the MSF catalog waits for, and if so -/// whether it carries a framerate. Text detects nothing and data tracks aren't listed in MSF. -fn msf_framerate() -> Option { - use std::any::TypeId; - let ty = TypeId::of::(); - if ty == TypeId::of::() { - Some(true) - } else if ty == TypeId::of::() { - Some(false) - } else { - None - } -} - -/// Whether `supplied` hints every field an MSF track's estimate would otherwise wait on. -fn hinted(supplied: &Estimate, framerate: bool) -> bool { - supplied.jitter.is_some() - && supplied.bitrate.is_some() - && supplied.delay.is_some() - && (!framerate || supplied.framerate.is_some()) -} - /// Refuse an edit that redefines rendition `name`'s track identity in place: a publisher whose /// decoder identity changed publishes under a new name instead. fn check_identity(previous: &Catalog, next: &Catalog, name: &str) -> crate::Result<()> { @@ -586,15 +565,13 @@ pub(crate) struct Rendition> { /// SPS) holds the handle without a catalog entry, and drops without a spurious removal. present: bool, /// The fields the config carried at [`set`](Self::set): authoritative, never overwritten by - /// detection. Whatever is absent here is what detection fills. + /// detection. Whatever is absent here is what detection fills, and stays absent from MSF. supplied: Estimate, /// The last estimate handed to [`estimate`](Self::estimate), kept so a config set afterwards /// starts from whatever was already measured. detected: Estimate, /// The last estimate written to the config, so an unchanged one doesn't republish the catalog. published: Option, - /// Whether the estimates the MSF catalog freezes have been made, so it no longer waits on them. - settled: bool, _config: PhantomData C>, } @@ -643,7 +620,6 @@ impl> Rendition { supplied: Estimate::default(), detected: Estimate::default(), published: None, - settled: false, _config: PhantomData, }) } @@ -672,9 +648,9 @@ impl> Rendition { /// /// Whatever [`Estimate`] fields `config` already carries are authoritative and left alone; the /// rest are filled by [`estimate`](Self::estimate), seeded with anything already measured before - /// the rendition existed (a dirty start or a B-frame reorder). A caller who wants to pre-empt - /// detection sets the field on the config, or (for a config an importer builds out of the - /// bitstream) hands the importer a hint like [`VideoHint`]. + /// the rendition existed (a dirty start or a B-frame reorder). Only the authoritative ones reach + /// the MSF catalog. A caller who wants to pre-empt detection sets the field on the config, or (for + /// a config an importer builds out of the bitstream) hands the importer a hint like [`VideoHint`]. pub(crate) fn set(&mut self, mut config: C) -> crate::Result<()> { let supplied = config.estimate(); let resolved = Self::resolved(&supplied, &self.detected); @@ -690,10 +666,9 @@ impl> Rendition { self.present = true; *guard = next; guard.list::(&self.name); - // Every estimated field hinted leaves nothing to wait for. - self.settled |= msf_framerate::().is_none_or(|framerate| hinted(&supplied, framerate)); - // A replaced config that hints everything releases the hold its predecessor took. - guard.unsettled::(&self.name, !self.settled); + if C::detects() { + guard.hint::(&self.name, supplied.clone()); + } guard.commit()?; } self.supplied = supplied; @@ -741,30 +716,6 @@ impl> Rendition { self.detected = estimate; return Ok(()); } - // The bitrate's first window closes on the first group boundary after a window of media. An - // unhinted framerate takes its own window after that, so MSF declares it absent. - let settles = !self.settled && estimate.bitrate.is_some(); - self.publish(estimate, period)?; - if settles { - self.settle()?; - } - Ok(()) - } - - /// Stop holding the MSF catalog back for this rendition's estimates: they are made, or the track - /// has ended and they never will be. - pub(crate) fn settle(&mut self) -> crate::Result<()> { - if std::mem::replace(&mut self.settled, true) || !self.present { - return Ok(()); - } - // A closed catalog has nothing left to publish. - if let Ok(guard) = self.catalog.modify() { - guard.settle::(&self.name)?; - } - Ok(()) - } - - fn publish(&mut self, estimate: Estimate, period: Option) -> crate::Result<()> { let mut resolved = Self::resolved(&self.supplied, &estimate); // A measurement never lowers the published jitter or delay, including one raised through // `modify`. @@ -839,8 +790,15 @@ impl> Rendition { let mut estimate = config.estimate(); estimate.bitrate = Some(bitrate); config.set_estimate(estimate); - self.replace(config)?; - self.supplied.bitrate = Some(bitrate); + let mut supplied = self.supplied.clone(); + supplied.bitrate = Some(bitrate); + let name = self.name.clone(); + let mut guard = self.stage(config)?; + if C::detects() { + guard.hint::(&name, supplied.clone()); + } + guard.commit()?; + self.supplied = supplied; Ok(()) } diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index 08c99758cdc..d4bc1eecfd5 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -105,13 +105,6 @@ impl Rendition { Self::Audio(rendition) => rendition.estimate(reading), } } - - fn settle(&mut self) -> crate::Result<()> { - match self { - Self::Video(rendition) => rendition.settle(), - Self::Audio(rendition) => rendition.settle(), - } - } } struct Fmp4Track { @@ -205,8 +198,6 @@ impl Passthrough { fn finish(&mut self) -> crate::Result<()> { self.close()?; self.track.finish()?; - // Its estimates are as made as they will be, so the MSF catalog stops waiting on them. - self.rendition.settle()?; // The importer may be kept around, but its media stops holding the other tracks back. self.estimator.discontinuity(); // Dropping the recorder stops the open HLS segment, so the track is advertised. diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 4309e8363eb..5a3c6c1835a 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -1381,9 +1381,6 @@ async fn test_msf_catalog_roundtrip() { let buf = bytes::BytesMut::from(&data[..]); // Trailing fragments may error out (e.g. partial mdat); ignore. let _ = fmp4.decode(&buf); - // The clip is shorter than a second, so its estimates are never made: the MSF catalog waits for - // the tracks to end. - fmp4.finish().unwrap(); let track = consumer .track(moq_msf::DEFAULT_NAME) diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index 2b4147df7f2..7a7e65e9817 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -108,7 +108,6 @@ trait Rendition: Send { fn config(&self) -> crate::Result; fn replace(&mut self, config: R) -> crate::Result<()>; fn estimate(&mut self, reading: crate::catalog::Reading) -> crate::Result<()>; - fn settle(&mut self) -> crate::Result<()>; fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()>; fn fresh( &self, @@ -161,10 +160,6 @@ where self.estimate(reading) } - fn settle(&mut self) -> crate::Result<()> { - self.settle() - } - fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()> { self.supply_bitrate(bitrate) } @@ -809,10 +804,6 @@ where self.cut(None)?; self.inner.finish()?; self.publish_estimate()?; - // Its estimates are as made as they will be, so the MSF catalog stops waiting on them. - if let Some(rendition) = self.rendition.as_mut() { - rendition.settle()?; - } // A finished track may be kept around, but its media stops holding the other tracks back. self.estimator.discontinuity(); // Dropping the recorder stops the open HLS segment, so a track that ends before its catalog