From ab7bab15923db99e98e21f09a42dfb4259e355ca Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:11:19 -0700 Subject: [PATCH 01/14] chore(quest): claim Rust catalog identity enforcement Existing approved batch slot11 transferred untouched from worker A to root. No additional agent or batch expansion. Co-Authored-By: GPT-6 From b42c1729d3f80942d8df205dd71b4b06a8c599eb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:23:26 -0700 Subject: [PATCH 02/14] test(mux): reproduce same-name media identity rewrites Audio and video identity edits currently succeed through the catalog setters; regressions require refusal without retaining the edit. Coded size, bitrate, and framerate controls remain allowed. Production unchanged; importer rotations and remaining publisher fixes are pending. Co-Authored-By: GPT-6 --- rs/moq-mux/src/catalog/tracks.rs | 88 ++++++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index 51ea9165b52..f991222a2d4 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -756,6 +756,94 @@ mod tests { config } + #[test] + fn video_identity_edits_are_refused_without_changing_the_catalog() { + let original = config(Some(1_000_000), None); + let mut changed = Vec::new(); + let mut codec = original.clone(); + codec.codec = "avc1.42c01e".parse().unwrap(); + changed.push(codec); + let mut description = original.clone(); + description.description = Some(bytes::Bytes::from_static(&[1, 2, 3])); + changed.push(description); + let mut container = original.clone(); + container.container = hang::catalog::Container::Loc; + changed.push(container); + let mut width = original.clone(); + width.display_aspect_width = Some(4); + changed.push(width); + let mut height = original.clone(); + height.display_aspect_height = Some(3); + changed.push(height); + for next in changed { + for operation in ["set", "replace", "stage"] { + let (_broadcast, catalog, mut rendition) = video_track(); + rendition.set(original.clone()).unwrap(); + let refused = match operation { + "set" => rendition.set(next.clone()).is_err(), + "replace" => rendition.replace(next.clone()).is_err(), + "stage" => rendition.stage(next.clone()).is_err(), + _ => unreachable!(), + }; + assert!(refused, "{operation} retained identity change {next:?}"); + assert_eq!(catalog.snapshot().video.renditions["v"], original); + } + } + } + + #[test] + fn video_size_and_estimates_can_change_in_place() { + let (_broadcast, catalog, mut rendition) = video_track(); + let mut original = config(Some(1_000_000), None); + original.coded_width = Some(1280); + original.coded_height = Some(720); + original.framerate = Some(60.0); + rendition.set(original.clone()).unwrap(); + let mut next = original; + next.coded_width = Some(1920); + next.coded_height = Some(1080); + next.bitrate = Some(500_000); + next.framerate = Some(30.0); + rendition.set(next.clone()).unwrap(); + assert_eq!(catalog.snapshot().video.renditions["v"], next); + } + + #[test] + fn audio_identity_edits_are_refused_without_changing_the_catalog() { + let original = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2); + let mut changed = Vec::new(); + let mut description = original.clone(); + description.description = Some(bytes::Bytes::from_static(&[1, 2, 3])); + changed.push(description); + let mut container = original.clone(); + container.container = hang::catalog::Container::Loc; + changed.push(container); + let mut sample_rate = original.clone(); + sample_rate.sample_rate = 44_100; + changed.push(sample_rate); + let mut channels = original.clone(); + channels.channel_count = 1; + changed.push(channels); + for next in changed { + for operation in ["set", "replace", "stage"] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap(); + let reserved = catalog.reserve(); + let mut rendition = reserved.init::("a").unwrap(); + drop(reserved); + rendition.set(original.clone()).unwrap(); + let refused = match operation { + "set" => rendition.set(next.clone()).is_err(), + "replace" => rendition.replace(next.clone()).is_err(), + "stage" => rendition.stage(next.clone()).is_err(), + _ => unreachable!(), + }; + assert!(refused, "{operation} retained identity change {next:?}"); + assert_eq!(catalog.snapshot().audio.renditions["a"], original); + } + } + } + #[test] fn rejected_jitter_is_not_retained() { let (_broadcast, catalog, mut rendition) = video_track(); From e281fde01bffc49f23e267822e1bcdc68ffd5e5a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:38:54 -0700 Subject: [PATCH 03/14] fix(mux): checkpoint immutable decoder identities and video rotation Refuse identity edits in set/replace/stage and rotate the five elementary video importers onto never-reused names. Size, bitrate, and frame-rate updates remain in place. Guard tests29/29, existing mux suite1392/1392, targeted rotation regression passed. Remaining: C helper scope choice, audio/TS/FLV/GST/video producer integration, docs and full final gates. This is an incomplete checkpoint. Co-Authored-By: GPT-6 --- rs/moq-mux/src/catalog/producer.rs | 5 ++ rs/moq-mux/src/catalog/tracks.rs | 46 ++++++++++++++++++ rs/moq-mux/src/codec/av1/import.rs | 5 +- rs/moq-mux/src/codec/h264/import.rs | 5 +- rs/moq-mux/src/codec/h265/import.rs | 4 +- rs/moq-mux/src/codec/video.rs | 71 ++++++++++++++++++++++++++-- rs/moq-mux/src/codec/vp8/import.rs | 4 +- rs/moq-mux/src/codec/vp9/import.rs | 4 +- rs/moq-mux/src/container/producer.rs | 35 ++++++++++++++ rs/moq-mux/src/error.rs | 3 ++ rs/moq-mux/src/timeline.rs | 5 ++ 11 files changed, 167 insertions(+), 20 deletions(-) diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 0b80c54512e..6735d57e2db 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -733,6 +733,11 @@ impl Producer { } } + /// Mint a media name without reusing any earlier track's content identity. + pub(crate) fn fresh_track(&self, info: moq_net::track::Info) -> crate::Result { + self.timeline.fresh_track(info) + } + /// Build the media [`container::Producer`](crate::container::Producer) for `track`, /// enrolling it in the broadcast's timelines so its groups are indexed. /// diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index f991222a2d4..aeb594c0ebc 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -353,6 +353,38 @@ impl RenditionConfig for hang::catalog::TextConfig { } } +/// Whether a video decoder can keep reading the same track under this config. +pub(crate) fn same_video_identity(previous: &hang::catalog::VideoConfig, next: &hang::catalog::VideoConfig) -> bool { + previous.codec == next.codec + && previous.description == next.description + && previous.container == next.container + && previous.display_aspect_width == next.display_aspect_width + && previous.display_aspect_height == next.display_aspect_height +} + +/// Whether an audio decoder can keep reading the same track under this config. +fn same_audio_identity(previous: &hang::catalog::AudioConfig, next: &hang::catalog::AudioConfig) -> bool { + previous.codec == next.codec + && previous.description == next.description + && previous.container == next.container + && previous.sample_rate == next.sample_rate + && previous.channel_count == next.channel_count +} + +fn check_identity(previous: &Catalog, next: &Catalog, name: &str) -> crate::Result<()> { + if let (Some(previous), Some(next)) = (previous.video.renditions.get(name), next.video.renditions.get(name)) + && !same_video_identity(previous, next) + { + return Err(crate::Error::IdentityChanged(name.to_string())); + } + if let (Some(previous), Some(next)) = (previous.audio.renditions.get(name), next.audio.renditions.get(name)) + && !same_audio_identity(previous, next) + { + return Err(crate::Error::IdentityChanged(name.to_string())); + } + Ok(()) +} + /// A clonable reservation context handed to importers so they declare their tracks up front. /// /// Made via [`Producer::reserve`]. While any `Reserved` clone is alive the track set may still @@ -522,6 +554,15 @@ pub(crate) type VideoTrack = Rendition; pub(crate) type AudioTrack = Rendition; /// A single text (caption/subtitle) track's catalog rendition. See [`Rendition`]. impl> Rendition { + pub(crate) fn fresh( + &self, + container: super::hang::Container, + info: moq_net::track::Info, + ) -> crate::Result> { + let track = self.catalog.fresh_track(info)?; + self.catalog.track(track, container, None) + } + fn new(reserved: Reserved, name: String) -> crate::Result { Self::owned(reserved.producer(), Some(reserved), name) } @@ -583,6 +624,7 @@ impl> Rendition { let mut guard = self.catalog.modify()?; let mut next = (*guard).clone(); config.insert(&mut next, &self.name); + check_identity(&guard, &next, &self.name)?; // Serialization must succeed before the reserved snapshot retains the edit. serde_json::to_writer(std::io::sink(), &next).map_err(moq_json::Error::from)?; self.present = true; @@ -699,6 +741,7 @@ impl> Rendition { let mut guard = self.catalog.modify()?; let mut next = (*guard).clone(); config.insert(&mut next, &self.name); + check_identity(&guard, &next, &self.name)?; serde_json::to_writer(std::io::sink(), &next).map_err(moq_json::Error::from)?; *guard = next; Ok(guard) @@ -812,6 +855,9 @@ mod tests { fn audio_identity_edits_are_refused_without_changing_the_catalog() { let original = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2); let mut changed = Vec::new(); + let mut codec = original.clone(); + codec.codec = hang::catalog::AudioCodec::Flac; + changed.push(codec); let mut description = original.clone(); description.description = Some(bytes::Bytes::from_static(&[1, 2, 3])); changed.push(description); diff --git a/rs/moq-mux/src/codec/av1/import.rs b/rs/moq-mux/src/codec/av1/import.rs index 017d89ff7e0..b002243eec7 100644 --- a/rs/moq-mux/src/codec/av1/import.rs +++ b/rs/moq-mux/src/codec/av1/import.rs @@ -167,10 +167,7 @@ impl Import { Ok(()) } - /// Apply a resolved config, updating the catalog rendition in place. - /// - /// A changed config just re-mirrors the rendition; there are no fixed tracks - /// to reject a reconfiguration. + /// Apply a resolved config, minting a track when the decoder identity changes. fn apply_config(&mut self, config: hang::catalog::VideoConfig) -> crate::Result<()> { self.catalog.publish(&mut self.track, config) } diff --git a/rs/moq-mux/src/codec/h264/import.rs b/rs/moq-mux/src/codec/h264/import.rs index b2fe8a31649..d3cdaf9a9bf 100644 --- a/rs/moq-mux/src/codec/h264/import.rs +++ b/rs/moq-mux/src/codec/h264/import.rs @@ -178,10 +178,7 @@ impl Import { Ok(()) } - /// Apply a resolved config, updating the catalog rendition in place. - /// - /// A changed config (new avcC, or a new inline SPS) just re-mirrors the - /// rendition; there are no fixed tracks to reject a reconfiguration. + /// Apply a resolved config, minting a track when the decoder identity changes. fn apply_config(&mut self, config: hang::catalog::VideoConfig) -> crate::Result<()> { self.catalog.publish(&mut self.track, config) } diff --git a/rs/moq-mux/src/codec/h265/import.rs b/rs/moq-mux/src/codec/h265/import.rs index 63aac840cc4..a7dc925cc1e 100644 --- a/rs/moq-mux/src/codec/h265/import.rs +++ b/rs/moq-mux/src/codec/h265/import.rs @@ -181,10 +181,8 @@ impl Import { Ok(()) } - /// Apply a resolved config, updating the catalog rendition in place. + /// Apply a resolved config, minting a track when the decoder identity changes. /// - /// A changed config just re-mirrors the rendition; there are no fixed tracks to reject a - /// reconfiguration. fn apply_config(&mut self, config: hang::catalog::VideoConfig) -> crate::Result<()> { self.catalog.publish(&mut self.track, config) } diff --git a/rs/moq-mux/src/codec/video.rs b/rs/moq-mux/src/codec/video.rs index 7a5a64fd672..2a1eb06c1a4 100644 --- a/rs/moq-mux/src/codec/video.rs +++ b/rs/moq-mux/src/codec/video.rs @@ -79,16 +79,81 @@ impl Catalog { } /// Overlay the hint onto `config` and publish it to `rendition`, unless it matches the last - /// publish. A changed config just re-mirrors the rendition; there are no fixed tracks to - /// reject a reconfiguration. + /// publish. Decoder identity changes finish the previous track and mint a fresh name. pub(crate) fn publish(&mut self, track: &mut Track, mut config: hang::catalog::VideoConfig) -> crate::Result<()> { self.hint.apply(&mut config); if self.last.as_ref() == Some(&config) { return Ok(()); } tracing::debug!(name = ?track.name(), ?config, "starting track"); - track.set(config.clone())?; + track.reconfigure(config.clone())?; self.last = Some(config); Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::container::Frame; + + #[test] + fn decoder_changes_mint_fresh_names_but_size_and_estimates_stay_in_place() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::<()>::new(&mut broadcast, Default::default()).unwrap(); + let raw = broadcast + .create_track("video", catalog.track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let mut track = catalog + .video( + raw, + crate::catalog::hang::Container::Legacy(crate::container::Kind::Video), + None, + ) + .unwrap(); + let mut publishing = Catalog::new(VideoHint::default()); + let mut config = hang::catalog::VideoConfig::new("avc1.42c01e".parse::().unwrap()); + config.description = Some(bytes::Bytes::from_static(b"first")); + publishing.publish(&mut track, config.clone()).unwrap(); + track + .write(Frame { + timestamp: moq_net::Timestamp::ZERO, + payload: bytes::Bytes::from_static(b"old"), + keyframe: true, + duration: None, + }) + .unwrap(); + let old = track.consume(); + config.coded_width = Some(640); + config.coded_height = Some(360); + config.bitrate = Some(500_000); + config.framerate = Some(30.0); + publishing.publish(&mut track, config.clone()).unwrap(); + assert_eq!(track.name(), "video"); + config.bitrate = Some(250_000); + config.framerate = Some(15.0); + publishing.publish(&mut track, config.clone()).unwrap(); + assert_eq!(track.name(), "video"); + config.description = Some(bytes::Bytes::from_static(b"second")); + publishing.publish(&mut track, config.clone()).unwrap(); + let second = track.name().to_string(); + assert_ne!(second, "video"); + assert!(!catalog.snapshot().video.renditions.contains_key("video")); + assert_eq!(catalog.snapshot().video.renditions[&second], config); + assert_eq!(old.latest(), Some(0)); + track + .write(Frame { + timestamp: moq_net::Timestamp::ZERO, + payload: bytes::Bytes::from_static(b"new"), + keyframe: true, + duration: None, + }) + .unwrap(); + assert_eq!(track.track().latest(), Some(0)); + config.description = Some(bytes::Bytes::from_static(b"first")); + publishing.publish(&mut track, config).unwrap(); + assert_ne!(track.name(), "video"); + assert_ne!(track.name(), second); + track.finish().unwrap(); + } +} diff --git a/rs/moq-mux/src/codec/vp8/import.rs b/rs/moq-mux/src/codec/vp8/import.rs index 0d6321b68a4..a8fd7a203cd 100644 --- a/rs/moq-mux/src/codec/vp8/import.rs +++ b/rs/moq-mux/src/codec/vp8/import.rs @@ -62,10 +62,8 @@ impl Import { Ok(()) } - /// Apply a resolved config, updating the catalog rendition in place. + /// Apply a resolved config, minting a track when the decoder identity changes. /// - /// A changed config just re-mirrors the rendition; there are no fixed tracks to reject a - /// reconfiguration. fn apply_config(&mut self, config: hang::catalog::VideoConfig) -> crate::Result<()> { self.catalog.publish(&mut self.track, config) } diff --git a/rs/moq-mux/src/codec/vp9/import.rs b/rs/moq-mux/src/codec/vp9/import.rs index c8732659436..04f46cdc2e3 100644 --- a/rs/moq-mux/src/codec/vp9/import.rs +++ b/rs/moq-mux/src/codec/vp9/import.rs @@ -62,10 +62,8 @@ impl Import { Ok(()) } - /// Apply a resolved config, updating the catalog rendition in place. + /// Apply a resolved config, minting a track when the decoder identity changes. /// - /// A changed config just re-mirrors the rendition; there are no fixed tracks to reject a - /// reconfiguration. fn apply_config(&mut self, config: hang::catalog::VideoConfig) -> crate::Result<()> { self.catalog.publish(&mut self.track, config) } diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index f2afb8b9991..c41e0c5150f 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -106,6 +106,11 @@ trait Rendition: Send { fn config(&self) -> crate::Result; fn replace(&mut self, config: R) -> crate::Result<()>; fn estimate(&mut self, estimate: crate::catalog::Estimate) -> crate::Result<()>; + fn fresh( + &self, + container: crate::catalog::hang::Container, + info: moq_net::track::Info, + ) -> crate::Result>; } impl Rendition for crate::catalog::tracks::Rendition @@ -117,6 +122,14 @@ where self.name() } + fn fresh( + &self, + container: crate::catalog::hang::Container, + info: moq_net::track::Info, + ) -> crate::Result> { + self.fresh(container, info) + } + fn timestamp(&self, hint: Option) -> crate::Result { self.timestamp(hint) } @@ -257,6 +270,28 @@ where } } +impl Producer { + /// Publish a decoder change under a fresh raw track name. + pub(crate) fn reconfigure(&mut self, config: hang::catalog::VideoConfig) -> crate::Result<()> { + let rendition = self.rendition.as_ref().ok_or(crate::Error::NotPublished)?; + let previous = match rendition.config() { + Ok(previous) => previous, + Err(crate::Error::NotPublished) => return self.set(config), + Err(err) => return Err(err), + }; + if crate::catalog::tracks::same_video_identity(&previous, &config) { + return self.set(config); + } + let container = crate::catalog::hang::Container::try_from(&config)?; + let mut next = rendition.fresh(container, self.inner.info().clone())?; + next.buffer_duration = self.buffer_duration; + next.set(config)?; + self.finish()?; + *self = next; + Ok(()) + } +} + /// A published rendition config edited in place. pub struct Guard<'a, R> { rendition: &'a mut dyn Rendition, diff --git a/rs/moq-mux/src/error.rs b/rs/moq-mux/src/error.rs index 7299a9f4b4a..c7f621e0e5b 100644 --- a/rs/moq-mux/src/error.rs +++ b/rs/moq-mux/src/error.rs @@ -257,6 +257,9 @@ pub enum Error { /// Another publisher instance replaced a followed broadcast, and stitching was off. #[error("another publisher instance replaced broadcast `{0}`; enable stitch to follow it as a program switch")] Replaced(String), + /// A published rendition changed fields that require a new decoder configuration. + #[error("rendition `{0}` changed identity; publish it under a new track name")] + IdentityChanged(String), } impl Error { diff --git a/rs/moq-mux/src/timeline.rs b/rs/moq-mux/src/timeline.rs index c7fc1bc6844..bf378d2c290 100644 --- a/rs/moq-mux/src/timeline.rs +++ b/rs/moq-mux/src/timeline.rs @@ -502,6 +502,11 @@ impl Timelines { } } + /// Mint a media name that this broadcast has never used. + pub(crate) fn fresh_track(&self, info: moq_net::track::Info) -> crate::Result { + Ok(self.broadcast.unique_track("/media", info)?) + } + /// Enroll the track `name`, its records cut by `config`. /// /// Creates its timeline track, named by [`hang::timeline::default_name`], on first enrollment. From 8682eb3d86de5f413c90fb279110bf021d0d1d0b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:47:21 -0700 Subject: [PATCH 04/14] test(c): reproduce manual catalog identity redefinition The proposed refusal expects existing MOQ_ERROR_MUX(-29), while unchanged C production returns success. Production behavior awaits the maintainer scope choice; no ABI changes. Co-Authored-By: GPT-6 --- rs/moq-c/src/test.rs | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index b4ce355b71a..c6dc95b9ae4 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -711,6 +711,38 @@ fn publish_video_config_replaces_its_own_rendition() { assert_eq!(moq_origin_close(origin), 0); } +/// Manual catalog metadata cannot redefine the decoder identity of an existing raw track. +#[test] +fn manual_video_identity_change_is_refused() { + let origin = id(moq_origin_create()); + let broadcast = publish_broadcast(origin, b"catalog-identity-refusal"); + let name = "authored"; + let codec = "vp8"; + let mut video = moq_video_config { + name: name.as_ptr().cast(), + name_len: name.len(), + label: std::ptr::null(), + label_len: 0, + codec: codec.as_ptr().cast(), + codec_len: codec.len(), + description: std::ptr::null(), + description_len: 0, + coded_width: 640, + coded_height: 360, + container: moq_container::default(), + }; + assert_eq!(unsafe { moq_publish_video_config(broadcast, &video) }, 0); + let next = "vp09.00.10.08"; + video.codec = next.as_ptr().cast(); + video.codec_len = next.len(); + assert_eq!( + unsafe { moq_publish_video_config(broadcast, &video) }, + crate::error::MOQ_ERROR_MUX + ); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn publish_catalog_config_null_pointer() { let origin = id(moq_origin_create()); From bad7c9412fe619783f42e54b0633d0117a4079d7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 00:05:03 -0700 Subject: [PATCH 05/14] fix(audio): publish retuned encoder bitrate in place Update the existing rendition from the live encoder configuration after a successful bitrate change, retaining its track and decoder identity. A regression covers the immediate catalog value and subsequent encoded packets. Co-Authored-By: GPT-6 --- rs/moq-audio/src/encode/producer.rs | 31 ++++++++++++++++++++++++++++- 1 file changed, 30 insertions(+), 1 deletion(-) diff --git a/rs/moq-audio/src/encode/producer.rs b/rs/moq-audio/src/encode/producer.rs index 85465c01af4..115309b6a2c 100644 --- a/rs/moq-audio/src/encode/producer.rs +++ b/rs/moq-audio/src/encode/producer.rs @@ -303,7 +303,10 @@ impl Producer { /// Retune the live encoder to `bitrate`. pub fn set_bitrate(&mut self, bitrate: moq_net::bandwidth::Rate) -> Result<(), Error> { - self.encoder.set_bitrate(bitrate) + self.encoder.set_bitrate(bitrate)?; + // Pin the configured target so encoded-byte estimates cannot replace it. + self.track.set(self.encoder.catalog())?; + Ok(()) } /// Re-anchor the timeline to the next frame's timestamp, dropping any @@ -1283,6 +1286,32 @@ mod tests { assert_eq!(second.track_name(), "1.opus"); } + #[tokio::test(start_paused = true)] + async fn retuning_updates_the_published_bitrate_in_place() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); + let options = Options { + track: Some("audio".into()), + ..Options::default() + }; + let mut producer = Producer::new(&mut broadcast, catalog.clone(), Input::default(), &options).unwrap(); + let before = catalog.snapshot().audio.renditions["audio"].clone(); + let bitrate = moq_net::bandwidth::Rate::from_bps(64_000); + producer.set_bitrate(bitrate).unwrap(); + let after = catalog.snapshot(); + assert_eq!(after.audio.renditions.len(), 1); + let mut expected = before; + expected.bitrate = Some(64_000); + assert_eq!(after.audio.renditions["audio"], expected); + assert_eq!(producer.track_name(), "audio"); + assert_eq!(producer.bitrate(), bitrate); + // Encoded-byte detection must not replace the newly configured target. + for timestamp in [0, 20_000, 40_000] { + producer.write(&pcm_frame(&vec![0.1; 1920], timestamp)).unwrap(); + } + assert_eq!(catalog.snapshot().audio.renditions["audio"].bitrate, Some(64_000)); + } + /// The rendition a producer publishes is the encoder's own catalog, so neither can drift. #[tokio::test] async fn publishes_the_encoder_catalog() { From 642c4cc1b24f151a7ba97f833e81f4419d506bb8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 00:05:05 -0700 Subject: [PATCH 06/14] fix(gst): preserve rendition identity through caps changes Keep compatible producers live, mint a fresh name when decoder inputs change, and retain finished resolved descriptions through renegotiation and graceful catalog closure. Unresolved reservations release their gates; explicit pad release and failure retain their existing retirement behavior. Co-Authored-By: GPT-6 --- rs/moq-gst/src/sink/pad.rs | 203 +++++++++++++++++++++++++++++++++---- 1 file changed, 183 insertions(+), 20 deletions(-) diff --git a/rs/moq-gst/src/sink/pad.rs b/rs/moq-gst/src/sink/pad.rs index b15dff8cd2a..e279daec59f 100644 --- a/rs/moq-gst/src/sink/pad.rs +++ b/rs/moq-gst/src/sink/pad.rs @@ -40,7 +40,7 @@ enum Sink { /// Keep finished rendition owners until a whole run's catalog has closed. #[must_use] pub(super) struct Closed { - track: Option, + tracks: Vec, } /// An audio or video pad, published through a codec importer. @@ -250,8 +250,11 @@ pub enum PushOutcome { pub struct Pad { /// The requested pad gates the first catalog even before CAPS can name its rendition. reserved: Option, + catalog: Option, wall: Option>, track: Option, + /// Finished identities remain described in replayed catalogs until the run ends. + retired: Vec, caps: Option, /// Set once a producer build rejects this pad's caps or bitstream; further buffers are dropped and /// the track stays finalized. Isolated to the pad, so the session and other pads keep going. The @@ -271,8 +274,10 @@ impl Pad { pub fn new() -> Self { Self { reserved: None, + catalog: None, wall: None, track: None, + retired: Vec::new(), caps: None, failed: false, state: PadState::NoSegment, @@ -307,7 +312,10 @@ impl Pad { } }; if !resolved { - self.finalize()?; + let retired = std::mem::take(&mut self.retired); + let result = self.finalize(); + self.retired = retired; + result?; } } Ok(()) @@ -327,7 +335,7 @@ impl Pad { self.failed } - /// (Re)build the producer when the pad's caps change, under `track` when the pad was given a name. + /// Keep a compatible producer, or publish changed decoder identity under a fresh name. /// A build failure invalidates only this pad; the caller keeps the session and other pads alive. /// Identical caps re-sent as a sticky event keep the live producer. pub(super) fn observe_caps( @@ -339,6 +347,38 @@ impl Pad { if self.failed || (self.track.is_some() && self.caps.as_deref() == Some(options.caps)) { return CapsOutcome::Unchanged; } + self.catalog = Some(catalog.clone()); + let same_identity = self + .caps + .as_ref() + .is_some_and(|caps| same_caps_identity(caps, options.caps)); + if self.track.is_some() && same_identity { + if let Some(Sink::Text(text)) = &mut self.track { + let result = text.producer.modify().and_then(|mut config| { + config.lang = options + .caps + .structure(0) + .expect("one caps structure") + .get::("language-code") + .ok(); + config.commit() + }); + if let Err(err) = result { + self.fail(); + return CapsOutcome::Failed(format!("{err:#}")); + } + } + self.caps = Some(options.caps.clone()); + return CapsOutcome::Unchanged; + } + let name = options + .requested + .filter(|_| self.caps.is_some() && !same_identity) + .map(|name| broadcast.unique_name(&format!(".{name}"))); + let options = ProducerOptions { + requested: name.as_deref().or(options.requested), + ..options + }; let reserved = self.reserved.take().unwrap_or_else(|| catalog.reserve()); match self.build(broadcast, reserved, options) { Ok(name) => CapsOutcome::Active(name), @@ -369,8 +409,10 @@ impl Pad { "encoder is only supported on audio and video pads, not {}", structure.name() ); - // Renegotiation: finalize the previous producer before replacing it (closed once, not abandoned). - self.finalize()?; + // Keep the finished identity's owner: archived groups still need its catalog description. + let (closed, result) = self.finish(); + self.retired = closed.tracks; + result?; // Opaque data has no codec importer and no catalog entry, so it never reaches the codec match. if structure.name() == "application/octet-stream" { // A generated name would leave the track unfindable: nothing advertises it. @@ -840,19 +882,78 @@ impl Pad { /// Finish once, retaining rendition owners so closing a run cannot publish their removal. pub(super) fn finish(&mut self) -> (Closed, Result) { self.reserved.take(); - let mut closed = Closed { - track: self.track.take(), - }; - let Some(track) = closed.track.as_mut() else { - return (closed, Ok(false)); + let mut track = self.track.take(); + // Unresolved importers still hold a reservation that gates the first catalog. + // They have no historical description to retain and must release that gate. + let retain = match (&track, &self.catalog) { + (Some(Sink::Media(media)), Some(catalog)) => { + let mut catalog = catalog.clone(); + match catalog.modify() { + Ok(current) if media.audio => current.audio.renditions.contains_key(&media.track.name()), + Ok(current) => current.video.renditions.contains_key(&media.track.name()), + Err(_) => false, // The catalog is already terminal. + } + } + _ => true, }; - let result = match track { - Sink::Media(media) => media.track.finish().map_err(anyhow::Error::from), - Sink::Text(text) => text.producer.finish().map_err(anyhow::Error::from), - Sink::Opaque(producer) => producer.clone().finish().map_err(anyhow::Error::from), + let result = match track.as_mut() { + Some(Sink::Media(media)) => media.track.finish().map_err(anyhow::Error::from), + Some(Sink::Text(text)) => text.producer.finish().map_err(anyhow::Error::from), + Some(Sink::Opaque(producer)) => producer.clone().finish().map_err(anyhow::Error::from), + None => { + return ( + Closed { + tracks: std::mem::take(&mut self.retired), + }, + Ok(false), + ); + } }; self.failed |= result.is_err(); - (closed, result.map(|()| true)) + let mut tracks = std::mem::take(&mut self.retired); + if retain { + tracks.extend(track); + } + (Closed { tracks }, result.map(|()| true)) + } +} + +/// Compare the decoder configuration this pad builds, ignoring unrelated caps hints. +fn same_caps_identity(previous: &gst::CapsRef, next: &gst::CapsRef) -> bool { + if previous.size() != 1 || next.size() != 1 { + return false; + } + let previous = previous.structure(0).expect("one caps structure"); + let next = next.structure(0).expect("one caps structure"); + if previous.name() != next.name() { + return false; + } + match previous.name().as_str() { + // Video configuration is in the encoded sequence headers. Its importer handles + // identity changes there; caps geometry, colour and rate fields are hints. + "video/x-h264" | "video/x-h265" | "video/x-av1" | "video/x-vp8" | "video/x-vp9" => true, + "audio/x-opus" => match (opus_catalog(previous), opus_catalog(next)) { + (Ok(previous), Ok(next)) => previous == next, + _ => false, + }, + "audio/mpeg" => { + let mp3 = |caps: &gst::StructureRef| caps.get::("layer").ok() == Some(3); + if mp3(previous) != mp3(next) { + return false; + } + if mp3(previous) { + return ["rate", "channels"].into_iter().all( + |field| matches!((previous.get::(field), next.get::(field)), (Ok(a), Ok(b)) if a > 0 && a == b), + ); + } + let description = |caps: &gst::StructureRef| { + let data = caps.get::("codec_data").ok()?; + Some(data.map_readable().ok()?.as_slice().to_vec()) + }; + matches!((description(previous), description(next)), (Some(a), Some(b)) if a == b) + } + "text/x-raw" | "application/octet-stream" => true, + _ => previous == next, } } @@ -1233,10 +1334,9 @@ mod tests { assert!(first.finalize().unwrap(), "the first producer is still live"); } - // Renegotiation re-reserves the same explicit name. `build` finalizes the old producer first and the - // broadcast reclaims closed entries on insert, so the pad does not collide with itself. + // A size hint does not change decoder identity or restart the existing track. #[test] - fn renegotiation_keeps_the_explicit_name() { + fn a_size_hint_keeps_the_live_producer() { gst::init().unwrap(); let (broadcast, catalog) = producers(); let mut pad = Pad::new(); @@ -1248,15 +1348,52 @@ mod tests { .field("stream-format", "byte-stream") .field("alignment", "au") .field("width", 1280i32) + .field("height", 720i32) + .field("framerate", gst::Fraction::new(24, 1)) + .field("colorimetry", "bt709") .build(); assert_eq!( pad.observe_caps(&broadcast, &catalog, producer_options(&renegotiated, Some("camera")),), - CapsOutcome::Active("camera".to_string()), - "the same name is reserved again, not rejected as a duplicate" + CapsOutcome::Unchanged, + "a size hint must not restart the track" ); assert!(!pad.is_failed()); } + #[test] + fn changed_audio_identity_mints_a_fresh_explicit_name() { + gst::init().unwrap(); + let (broadcast, catalog) = producers(); + let mut pad = Pad::new(); + assert_eq!( + pad.observe_caps( + &broadcast, + &catalog, + producer_options(&opus_caps(1, None, None), Some("audio")) + ), + CapsOutcome::Active("audio".into()) + ); + let first = catalog.snapshot().audio.renditions["audio"].clone(); + let CapsOutcome::Active(next) = pad.observe_caps( + &broadcast, + &catalog, + producer_options(&opus_caps(2, None, None), Some("audio")), + ) else { + panic!("new audio identity must publish"); + }; + assert_ne!(next, "audio"); + let after = catalog.snapshot(); + assert_eq!(after.audio.renditions["audio"], first); + assert_eq!(after.audio.renditions[&next].channel_count, 2); + let (closed, result) = pad.finish(); + assert!(result.unwrap()); + drop(pad); + assert_eq!(catalog.snapshot().audio.renditions, after.audio.renditions); + let mut final_catalog = catalog.clone(); + final_catalog.finish().unwrap(); + drop(closed); + } + // AAC carries its config in caps; without codec_data the producer cannot be built. #[test] fn aac_without_codec_data_fails_the_pad() { @@ -2105,6 +2242,32 @@ mod tests { assert!(!pad.finalize().unwrap(), "no producer to finalize"); } + #[test] + fn subtitle_language_updates_in_place() { + gst::init().unwrap(); + let (broadcast, catalog) = producers(); + let mut pad = Pad::new(); + let caps = gst::Caps::builder("text/x-raw") + .field("format", "utf8") + .field("language-code", "en") + .build(); + assert_eq!( + pad.observe_caps(&broadcast, &catalog, producer_options(&caps, Some("subtitles"))), + CapsOutcome::Active("subtitles".into()) + ); + let next = gst::Caps::builder("text/x-raw") + .field("format", "utf8") + .field("language-code", "fr") + .build(); + assert_eq!( + pad.observe_caps(&broadcast, &catalog, producer_options(&next, Some("subtitles"))), + CapsOutcome::Unchanged + ); + let after = catalog.snapshot(); + assert_eq!(after.text.renditions.len(), 1); + assert_eq!(after.text.renditions["subtitles"].lang.as_deref(), Some("fr")); + } + fn text_caps() -> gst::Caps { gst::Caps::builder("text/x-raw").field("format", "utf8").build() } From 9208df3b547afa1e68ddbc2959b8eb14d01acb3a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 09:13:47 -0700 Subject: [PATCH 07/14] fix(mux)!: mint a new track for every Rust decoder identity change hang::catalog::{VideoConfig, AudioConfig}::same_identity name the draft's identity fields; moq-mux refuses an in-place redefinition and moq-c refuses an authored one. TS AAC/MP2/AC-3 and FLV in-band audio start a new track when a frame header changes the format, and moq-video's capture follows a reopen onto a new track. Co-Authored-By: Claude Opus 5.5 --- rs/hang/src/catalog/audio/mod.rs | 15 +++ rs/hang/src/catalog/video/mod.rs | 16 +++ rs/moq-c/src/api.rs | 14 ++- rs/moq-c/src/publish.rs | 15 ++- rs/moq-c/src/test.rs | 36 ++++++ rs/moq-ffi/src/video.rs | 4 +- rs/moq-mux/src/catalog/tracks.rs | 25 +--- rs/moq-mux/src/codec/h264/import.rs | 9 +- rs/moq-mux/src/codec/h265/import.rs | 9 +- rs/moq-mux/src/codec/video.rs | 2 +- rs/moq-mux/src/container/flv/import.rs | 31 +++-- rs/moq-mux/src/container/flv/import_test.rs | 42 +++++++ rs/moq-mux/src/container/producer.rs | 2 +- rs/moq-mux/src/container/ts/import.rs | 127 +++++++++++++++++++- rs/moq-video/src/encode/producer.rs | 121 ++++++++++++++----- 15 files changed, 381 insertions(+), 87 deletions(-) diff --git a/rs/hang/src/catalog/audio/mod.rs b/rs/hang/src/catalog/audio/mod.rs index 0c13373888d..49bdda97753 100644 --- a/rs/hang/src/catalog/audio/mod.rs +++ b/rs/hang/src/catalog/audio/mod.rs @@ -235,6 +235,21 @@ impl AudioConfig { hls: None, } } + + /// Whether `other` keeps this config's track identity: the fields a consumer cannot change + /// without reinitializing its decoder or demuxer (`codec`, `description`, `container`, + /// `sample_rate`, and `channel_count`). + /// + /// A track name always means the same content, so a publisher that changes any of them + /// publishes under a new name; every other field may update in place. See the draft's + /// [Track Identity](https://github.com/moq-dev/moq/blob/main/drafts/draft-lcurley-moq-hang.md#identity). + pub fn same_identity(&self, other: &Self) -> bool { + self.codec == other.codec + && self.description == other.description + && self.container == other.container + && self.sample_rate == other.sample_rate + && self.channel_count == other.channel_count + } } #[cfg(test)] diff --git a/rs/hang/src/catalog/video/mod.rs b/rs/hang/src/catalog/video/mod.rs index 758a8d2bff7..03693ae9eb5 100644 --- a/rs/hang/src/catalog/video/mod.rs +++ b/rs/hang/src/catalog/video/mod.rs @@ -394,6 +394,22 @@ impl VideoConfig { hls: None, } } + + /// Whether `other` keeps this config's track identity: the fields a consumer cannot change + /// without reinitializing its decoder or demuxer (`codec`, `description`, `container`, + /// `display_aspect_width`, `display_aspect_height`, and `interlaced`). + /// + /// A track name always means the same content, so a publisher that changes any of them + /// publishes under a new name; every other field may update in place. See the draft's + /// [Track Identity](https://github.com/moq-dev/moq/blob/main/drafts/draft-lcurley-moq-hang.md#identity). + pub fn same_identity(&self, other: &Self) -> bool { + self.codec == other.codec + && self.description == other.description + && self.container == other.container + && self.display_aspect_width == other.display_aspect_width + && self.display_aspect_height == other.display_aspect_height + && self.interlaced == other.interlaced + } } fn is_zero(value: &i32) -> bool { diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 603979dfcef..b93ea8db0eb 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -2613,10 +2613,12 @@ pub unsafe extern "C" fn moq_publish_video_properties(broadcast: u32, properties /// reading a rendition out of a catalog, it writes one into the catalog of a /// broadcast created with [moq_origin_create_broadcast]. The rendition is keyed by /// `config.name`; calling this again with the same name replaces the rendition -/// you declared, so a config can be refined in place. It fails only when a +/// you declared, so a config can be refined in place. It fails when a /// [moq_publish_video] track owns the name, since that track publishes and -/// retires its own rendition. The updated catalog is published to subscribers -/// automatically. +/// retires its own rendition, and when the replacement changes the track's +/// identity (`codec`, `description`, or `container`), since a track name always +/// means the same content: publish the changed media under a new name instead. +/// The updated catalog is published to subscribers automatically. /// /// The struct fields are read as inputs: /// - `name` / `codec` are required (NOT NULL terminated) string slices. @@ -2663,8 +2665,10 @@ pub unsafe extern "C" fn moq_publish_video_config(broadcast: u32, config: *const /// /// This is the producer counterpart to [moq_consume_audio_config]. The rendition /// is keyed by `config.name`, on the same terms as [moq_publish_video_config]: -/// a repeat call replaces your own rendition, and a name a [moq_publish_audio] -/// track owns is refused. The updated catalog is published to subscribers +/// a repeat call replaces your own rendition, a name a [moq_publish_audio] +/// track owns is refused, and so is a replacement that changes the track's +/// identity (`codec`, `description`, `container`, `sample_rate`, or +/// `channel_count`). The updated catalog is published to subscribers /// automatically. /// /// The struct fields are read as inputs: diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index a73a705cff6..2f6f6b71ba9 100644 --- a/rs/moq-c/src/publish.rs +++ b/rs/moq-c/src/publish.rs @@ -672,13 +672,19 @@ impl Publish { /// Insert or replace a caller-authored video rendition in the broadcast's catalog. /// - /// Errors if a media importer owns the name, since it publishes and retires its own rendition. - /// The catalog is republished automatically. + /// Errors if a media importer owns the name, since it publishes and retires its own rendition, + /// or if the config changes the rendition's track identity, which takes a new name. The catalog + /// is republished automatically. pub fn video_config(&mut self, broadcast: Id, name: &str, config: hang::catalog::VideoConfig) -> Result<(), Error> { let broadcast = self.broadcast_mut(broadcast)?; if !broadcast.video.contains_key(name) && broadcast.catalog.is_claimed::(name) { return Err(Error::Hang(hang::Error::Duplicate(name.to_string()))); } + if let Some(previous) = broadcast.video.get(name) + && !previous.same_identity(&config) + { + return Err(moq_mux::Error::IdentityChanged(name.to_string()).into()); + } broadcast.video.insert(name.to_string(), config.clone()); let mut catalog = broadcast.catalog.modify()?; catalog.video.renditions.insert(name.to_string(), config); @@ -694,6 +700,11 @@ impl Publish { if !broadcast.audio.contains_key(name) && broadcast.catalog.is_claimed::(name) { return Err(Error::Hang(hang::Error::Duplicate(name.to_string()))); } + if let Some(previous) = broadcast.audio.get(name) + && !previous.same_identity(&config) + { + return Err(moq_mux::Error::IdentityChanged(name.to_string()).into()); + } broadcast.audio.insert(name.to_string(), config.clone()); let mut catalog = broadcast.catalog.modify()?; catalog.audio.renditions.insert(name.to_string(), config); diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index d70d6991a68..7b336585d07 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -827,6 +827,7 @@ fn manual_video_identity_change_is_refused() { coded_width: 640, coded_height: 360, container: moq_container::default(), + preference: 0, }; assert_eq!(unsafe { moq_publish_video_config(broadcast, &video) }, 0); let next = "vp09.00.10.08"; @@ -840,6 +841,41 @@ fn manual_video_identity_change_is_refused() { assert_eq!(moq_origin_close(origin), 0); } +/// Manual catalog metadata cannot redefine the decoder identity of an existing audio track, while +/// a live field such as the label still updates in place. +#[test] +fn manual_audio_identity_change_is_refused() { + let origin = id(moq_origin_create()); + let broadcast = publish_broadcast(origin, b"catalog-audio-identity-refusal"); + let name = "authored"; + let codec = "opus"; + let label = "English"; + let mut audio = moq_audio_config { + name: name.as_ptr().cast(), + name_len: name.len(), + label: std::ptr::null(), + label_len: 0, + codec: codec.as_ptr().cast(), + codec_len: codec.len(), + description: std::ptr::null(), + description_len: 0, + sample_rate: 48_000, + channel_count: 2, + container: moq_container::default(), + }; + assert_eq!(unsafe { moq_publish_audio_config(broadcast, &audio) }, 0); + audio.label = label.as_ptr().cast(); + audio.label_len = label.len(); + assert_eq!(unsafe { moq_publish_audio_config(broadcast, &audio) }, 0); + audio.channel_count = 1; + assert_eq!( + unsafe { moq_publish_audio_config(broadcast, &audio) }, + crate::error::MOQ_ERROR_MUX + ); + assert_eq!(moq_publish_close(broadcast), 0); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn publish_catalog_config_null_pointer() { let origin = id(moq_origin_create()); diff --git a/rs/moq-ffi/src/video.rs b/rs/moq-ffi/src/video.rs index a50fe16113f..f837d2b9d4f 100644 --- a/rs/moq-ffi/src/video.rs +++ b/rs/moq-ffi/src/video.rs @@ -1804,7 +1804,9 @@ mod rung_tests { ladder.high.write(frame(i)).unwrap(); } ladder.high.cut().unwrap(); - ladder.high.set_bitrate(3_000_000).unwrap(); + // Within the level the rung declared: a ceiling past it reopens the encoder at a new + // level, which is a new codec string and so a new track. + ladder.high.set_bitrate(1_500_000).unwrap(); apply(Target::Enabled(Rate::from_kbps(1_500))); assert!(enabled(&ladder.broadcast, "high")); diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index b3a049becff..acd6f4fd21a 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -367,33 +367,16 @@ impl RenditionConfig for hang::catalog::TextConfig { } } -/// Whether a video decoder can keep reading the same track under this config. -pub(crate) fn same_video_identity(previous: &hang::catalog::VideoConfig, next: &hang::catalog::VideoConfig) -> bool { - previous.codec == next.codec - && previous.description == next.description - && previous.container == next.container - && previous.display_aspect_width == next.display_aspect_width - && previous.display_aspect_height == next.display_aspect_height - && previous.interlaced == next.interlaced -} - -/// Whether an audio decoder can keep reading the same track under this config. -fn same_audio_identity(previous: &hang::catalog::AudioConfig, next: &hang::catalog::AudioConfig) -> bool { - previous.codec == next.codec - && previous.description == next.description - && previous.container == next.container - && previous.sample_rate == next.sample_rate - && previous.channel_count == next.channel_count -} - +/// 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<()> { if let (Some(previous), Some(next)) = (previous.video.renditions.get(name), next.video.renditions.get(name)) - && !same_video_identity(previous, next) + && !previous.same_identity(next) { return Err(crate::Error::IdentityChanged(name.to_string())); } if let (Some(previous), Some(next)) = (previous.audio.renditions.get(name), next.audio.renditions.get(name)) - && !same_audio_identity(previous, next) + && !previous.same_identity(next) { return Err(crate::Error::IdentityChanged(name.to_string())); } diff --git a/rs/moq-mux/src/codec/h264/import.rs b/rs/moq-mux/src/codec/h264/import.rs index c90872aef2b..407420cd792 100644 --- a/rs/moq-mux/src/codec/h264/import.rs +++ b/rs/moq-mux/src/codec/h264/import.rs @@ -122,7 +122,8 @@ impl Import { Ok(()) } - /// The MoQ track name this importer publishes on. + /// The MoQ track name this importer publishes on, which a decoder identity change replaces + /// with a newly minted one. pub fn name(&self) -> &str { self.track.name() } @@ -132,6 +133,12 @@ impl Import { self.track.track().demand() } + /// The track this importer publishes on, which a decoder identity change replaces with a + /// newly named one. + pub fn track(&self) -> &moq_net::track::Producer { + self.track.track() + } + /// Finish the track, flushing any buffered data. pub fn finish(&mut self) -> Result<()> { self.track.finish()?; diff --git a/rs/moq-mux/src/codec/h265/import.rs b/rs/moq-mux/src/codec/h265/import.rs index 4e77af2cbfe..121f157fa6a 100644 --- a/rs/moq-mux/src/codec/h265/import.rs +++ b/rs/moq-mux/src/codec/h265/import.rs @@ -125,7 +125,8 @@ impl Import { Ok(()) } - /// The MoQ track name this importer publishes on. + /// The MoQ track name this importer publishes on, which a decoder identity change replaces + /// with a newly minted one. pub fn name(&self) -> &str { self.track.name() } @@ -135,6 +136,12 @@ impl Import { self.track.track().demand() } + /// The track this importer publishes on, which a decoder identity change replaces with a + /// newly named one. + pub fn track(&self) -> &moq_net::track::Producer { + self.track.track() + } + /// Finish the track, flushing the current group. pub fn finish(&mut self) -> Result<()> { self.track.finish()?; diff --git a/rs/moq-mux/src/codec/video.rs b/rs/moq-mux/src/codec/video.rs index b838ce5b691..28e5a15cd6a 100644 --- a/rs/moq-mux/src/codec/video.rs +++ b/rs/moq-mux/src/codec/video.rs @@ -260,7 +260,7 @@ impl Catalog { let declared = self .last .as_ref() - .filter(|last| crate::catalog::tracks::same_video_identity(last, &config)) + .filter(|last| last.same_identity(&config)) .and_then(|last| last.reorder); if let Some(declared) = declared { match config.reorder { diff --git a/rs/moq-mux/src/container/flv/import.rs b/rs/moq-mux/src/container/flv/import.rs index 49780014e42..33da5304fb3 100644 --- a/rs/moq-mux/src/container/flv/import.rs +++ b/rs/moq-mux/src/container/flv/import.rs @@ -16,7 +16,7 @@ //! Each codec's out-of-band config record (avcC / hvcC / av1C / `AudioSpecificConfig` //! / `OpusHead`) becomes the catalog `description`; VP9 and the verbatim audio //! codecs (MP3 / AC-3 / E-AC-3) carry their config in band, so they configure from -//! the first frame instead. Sample bytes already match the codec payload, so no codec +//! the frames instead, a changed config starting a new track. Sample bytes already match the codec payload, so no codec //! transform is needed before the selected media container wraps them. FLAC (`fLaC`) //! enhanced audio, and any other unsupported codec, fail the import. @@ -349,11 +349,10 @@ impl Import { } if sound_format == AUDIO_FORMAT_MP3 { // Legacy MP3: the raw frame follows the one-byte tag header, with the - // config in band. Configure from the first frame, then write it. + // config in band. Configure from each frame, which starts a new track only + // when the header changes, then write it. let frame = &body[1..]; - if !self.audio.contains_key(&DEFAULT_TRACK_ID) { - self.init_audio(DEFAULT_TRACK_ID, config_from_mp3(frame)?)?; - } + self.init_audio(DEFAULT_TRACK_ID, config_from_mp3(frame)?)?; return self.write_audio(DEFAULT_TRACK_ID, frame, timestamp); } if sound_format != AUDIO_FORMAT_AAC { @@ -437,18 +436,16 @@ impl Import { self.init_audio(track_id, config) } AUDIO_PACKET_CODED_FRAMES => { - // MP3 / AC-3 / E-AC-3 carry their config in the frame header, so - // configure from the first frame when no sequence header preceded it. - if !self.audio.contains_key(&track_id) { - let config = match fourcc { - b".mp3" => Some(config_from_mp3(payload)?), - b"ac-3" => Some(config_from_ac3(payload)?), - b"ec-3" => Some(config_from_eac3(payload)?), - _ => None, - }; - if let Some(config) = config { - self.init_audio(track_id, config)?; - } + // MP3 / AC-3 / E-AC-3 carry their config in the frame header, so configure from + // each frame, which starts a new track only when the header changes. + let config = match fourcc { + b".mp3" => Some(config_from_mp3(payload)?), + b"ac-3" => Some(config_from_ac3(payload)?), + b"ec-3" => Some(config_from_eac3(payload)?), + _ => None, + }; + if let Some(config) = config { + self.init_audio(track_id, config)?; } self.write_audio(track_id, payload, timestamp) } diff --git a/rs/moq-mux/src/container/flv/import_test.rs b/rs/moq-mux/src/container/flv/import_test.rs index 9c68a77a074..c84abeb5ee7 100644 --- a/rs/moq-mux/src/container/flv/import_test.rs +++ b/rs/moq-mux/src/container/flv/import_test.rs @@ -341,6 +341,48 @@ async fn import_legacy_mp3() { assert!(a.description.is_none(), "MP3 config is in band"); } +/// A legacy MP3 frame header that changes the channel count is a new decoder identity, so the +/// frames after it publish on a new track rather than rewriting the first. +#[tokio::test(start_paused = true)] +async fn legacy_mp3_channel_change_starts_a_new_track() { + // MPEG-1 Layer III, 128 kbps, 44.1 kHz: joint stereo, then mono. + let frame = |mode: u8| { + let mut mp3 = vec![0xFF, 0xFB, 0x90, mode]; + mp3.resize(417, 0xAA); + let mut tag = vec![super::MP3_AUDIO_TAG_HEADER]; + tag.extend_from_slice(&mp3); + tag + }; + let header = || { + let mut out = Vec::new(); + out.extend_from_slice(b"FLV"); + out.push(1); + out.push(0x04); // audio only + out.extend_from_slice(&9u32.to_be_bytes()); + out.extend_from_slice(&0u32.to_be_bytes()); + out + }; + + let mut producer = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut producer, crate::catalog::Config::default()).unwrap(); + let mut importer = Import::new(producer, catalog.reserve()); + let mut out = header(); + write_tag(&mut out, super::TAG_AUDIO, 0, &frame(0x44)); + write_tag(&mut out, super::TAG_AUDIO, 26, &frame(0x44)); + importer.decode(&bytes::BytesMut::from(out.as_slice())).unwrap(); + let first = catalog.snapshot().audio.renditions.keys().next().unwrap().clone(); + + let mut out = Vec::new(); + write_tag(&mut out, super::TAG_AUDIO, 52, &frame(0xC4)); + importer.decode(&bytes::BytesMut::from(out.as_slice())).unwrap(); + let snap = catalog.snapshot(); + assert_eq!(snap.audio.renditions.len(), 1, "the stereo track is retired"); + let (name, a) = snap.audio.renditions.iter().next().unwrap(); + assert_ne!(name, &first, "the mono frames publish under a new name"); + assert_eq!(a.channel_count, 1); + importer.finish().unwrap(); +} + /// A minimal avcC like [`avcc`], but with a caller-chosen level byte so two video /// renditions carry distinct codec configs. fn avcc_level(level: u8) -> Vec { diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index ff7d81a69cc..04470fae928 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -276,7 +276,7 @@ impl Producer { Err(crate::Error::NotPublished) => return self.set(config), Err(err) => return Err(err), }; - if crate::catalog::tracks::same_video_identity(&previous, &config) { + if previous.same_identity(&config) { return self.set(config); } let container = crate::catalog::hang::Container::try_from(&config)?; diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 4b48737537c..408a8ea3baa 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -894,9 +894,10 @@ impl Import { Some(StreamType::AdtsAac) => Stream::Aac(Box::new(AacStream { import: None, asc: bytes::Bytes::new(), + format: (0, 0, 0), broadcast: self.broadcast.clone(), reserved: Some(self.reserve()), - catalog: self.catalog.clone(), + timebase: self.timebase.clone(), container: self.container.clone(), tail: Vec::new(), tail_pts: None, @@ -988,6 +989,8 @@ impl Import { import: None, broadcast: self.broadcast.clone(), reserved: Some(self.reserve()), + timebase: self.timebase.clone(), + format: (0, 0), container: self.container.clone(), tail: Vec::new(), tail_pts: None, @@ -3645,13 +3648,17 @@ struct AacStream { /// The AudioSpecificConfig `import` was built with. A program config element in a later frame /// must rebuild it exactly. asc: bytes::Bytes, + /// The ADTS object type, sample rate, and channel configuration `import` was built with. A + /// frame header that changes them starts a new track. + format: (u8, u32, u8), broadcast: moq_net::broadcast::Producer, /// Reservation held from the PMT until the first frame builds the importer, so the catalog stays /// withheld until this deferred rendition resolves (config comes from the first ADTS header). /// Consumed when `import` is built, or released by a frame that cannot build it. reserved: Option>, - /// Reserves the rendition anew when `import` is built after `reserved` was released. - catalog: crate::catalog::Producer, + /// Reserves the rendition anew on the program's timebase when `import` is built after `reserved` + /// was released, or for the next track after a config change. + timebase: crate::catalog::Timebase, /// The container this importer publishes decoded renditions with. container: hang::catalog::Container, /// Partial frame left at the end of the previous PES. ISO 13818-1 doesn't require @@ -3791,6 +3798,16 @@ impl AacStream { }; let mut block = &data[offset + header.header_len..end]; + let format = (header.object_type, header.sample_rate, header.channel_config); + // A new object type, sample rate, or channel configuration is a new decoder identity: + // end this track, and publish the rest under a new name. + if let Some(import) = &mut self.import + && format != self.format + { + tracing::info!(name = import.name(), "AAC format changed mid-stream, starting a new track"); + import.finish()?; + self.import = None; + } let import = match &mut self.import { Some(import) => { // The description already carries the layout, so a repeated program config @@ -3832,12 +3849,13 @@ impl AacStream { config.container = self.container.clone(); // Consume the reservation held since the PMT: this resolves the gated rendition, // and carries the catalog's declared media retention onto the track. - let reserved = self.reserved.take().unwrap_or_else(|| self.catalog.reserve()); + let reserved = self.reserved.take().unwrap_or_else(|| self.timebase.reserve()); let track = self .broadcast .unique_track(".aac", reserved.track_info(hang::catalog::PRIORITY.audio))?; let aac = aac::Import::new(track, reserved, config)?; self.asc = asc; + self.format = format; self.import.insert(aac) } }; @@ -4262,6 +4280,10 @@ struct LegacyStream { /// withheld until this deferred rendition resolves (config comes from the first frame header). /// Consumed when `import` is built. reserved: Option>, + /// Reserves the next track on the program's timebase after a config change. + timebase: crate::catalog::Timebase, + /// The sample rate and channel count `import` was built with, from the first frame header. + format: (u32, u32), /// The container this importer publishes decoded renditions with. container: hang::catalog::Container, /// Partial frame left at the end of the previous PES. ISO 13818-1 doesn't @@ -4413,6 +4435,16 @@ impl LegacyStream { } }; + let format = (header.sample_rate, header.channel_count); + // A new sample rate or channel count is a new decoder identity: end this track, and + // publish the rest under a new name. + if let Some(import) = &mut self.import + && format != self.format + { + tracing::info!(name = import.name(), "audio format changed mid-stream, starting a new track"); + import.finish()?; + self.import = None; + } let import = match &mut self.import { Some(import) => import, None => { @@ -4422,13 +4454,15 @@ impl LegacyStream { container: self.container.clone(), }; // Consume the reservation held since the PMT: this resolves the gated rendition, - // and carries the catalog's declared media retention onto the track. - let reserved = self.reserved.take().expect("legacy reservation already consumed"); + // and carries the catalog's declared media retention onto the track. A track + // after a format change reserves anew on the same timebase. + let reserved = self.reserved.take().unwrap_or_else(|| self.timebase.reserve()); let track = self.broadcast.unique_track( self.descriptor.track_suffix, reserved.track_info(hang::catalog::PRIORITY.audio), )?; let legacy = legacy::Import::new(self.descriptor, track, reserved, config)?; + self.format = format; self.import.insert(legacy) } }; @@ -6300,6 +6334,87 @@ pub(super) mod test { f } + // A frame header that changes the channel count is a new decoder identity: the frames after + // it publish on a new track, and the first track is retired rather than rewritten. + #[tokio::test(start_paused = true)] + async fn legacy_channel_change_starts_a_new_track() { + const MP2_PID: u16 = 0x0061; + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + import + .decode(&synth_pmt(&[(StreamType::Mpeg1Audio, MP2_PID)], false)) + .unwrap(); + + // The same frame in stereo: the mode bits clear. + let stereo = |fill| { + let mut frame = mp2_frame(fill); + frame[3] = 0x00; + frame + }; + let mono = [mp2_frame(0xAA), mp2_frame(0xBB)].concat(); + import.decode(&audio_pes_packet(MP2_PID, 0, 90_000, &mono)).unwrap(); + let first = catalog.snapshot().audio.renditions.keys().next().unwrap().clone(); + let both = [stereo(0xCC), stereo(0xDD)].concat(); + import.decode(&audio_pes_packet(MP2_PID, 1, 180_000, &both)).unwrap(); + import.decode(&audio_pes_packet(MP2_PID, 2, 270_000, &both)).unwrap(); + + let snapshot = catalog.snapshot(); + assert_eq!(snapshot.audio.renditions.len(), 1, "the mono track is retired"); + let (name, config) = snapshot.audio.renditions.iter().next().unwrap(); + assert_ne!(name, &first, "the stereo frames publish under a new name"); + assert_eq!(config.channel_count, 2); + import.finish().unwrap(); + let frames = read_audio_frames(&consumer, &catalog).await; + assert_eq!(frames[0].payload, stereo(0xCC), "the new track starts at the change"); + } + + // An ADTS header that changes the sample rate is a new decoder identity, so the frames after + // it publish on a new track, on the same program timebase. + #[tokio::test(start_paused = true)] + async fn aac_sample_rate_change_starts_a_new_track() { + const AAC_PID: u16 = 0x0060; + let frame = |rate| { + let mut f = super::adts::write_header(2, rate, 2, 10).unwrap().to_vec(); + f.resize(f.len() + 10, 0x20); + f + }; + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.reserve()); + let mut mux = Mux { + out: synth_pmt(&[(StreamType::AdtsAac, AAC_PID)], false), + ..Default::default() + }; + let cc = mux.cc(AAC_PID); + mux.out.extend(audio_pes_packet(AAC_PID, cc, 90_000, &[frame(48_000), frame(48_000)].concat())); + import.decode(&mux.out).unwrap(); + let first = catalog.snapshot().audio.renditions.keys().next().unwrap().clone(); + + mux.out.clear(); + for pts in [180_000, 270_000] { + let cc = mux.cc(AAC_PID); + mux.out.extend(audio_pes_packet(AAC_PID, cc, pts, &[frame(44_100), frame(44_100)].concat())); + } + import.decode(&mux.out).unwrap(); + let snapshot = catalog.snapshot(); + assert_eq!(snapshot.audio.renditions.len(), 1, "the 48 kHz track is retired"); + let (name, config) = snapshot.audio.renditions.iter().next().unwrap(); + assert_ne!(name, &first, "the 44.1 kHz frames publish under a new name"); + assert_eq!(config.sample_rate, 44_100); + import.finish().unwrap(); + let frames = read_audio_frames(&consumer, &catalog).await; + assert_eq!( + frames[0].timestamp, + moq_net::Timestamp::from_micros(2_000_000).unwrap(), + "the new track keeps the program's timebase" + ); + } + // The shape a looping mux actually produces, which is not a carried tail: the last PES // before the cut is truncated, so it stays open, and the next loop's leading continuation // packets complete it. The foreign bytes land in the SAME PES, so the codec sees one diff --git a/rs/moq-video/src/encode/producer.rs b/rs/moq-video/src/encode/producer.rs index 3acba2f2f75..35e6148ea08 100644 --- a/rs/moq-video/src/encode/producer.rs +++ b/rs/moq-video/src/encode/producer.rs @@ -91,12 +91,6 @@ enum Codecs { /// handles catalog registration and framing. pub struct Producer { codecs: Codecs, - /// The published track, read for its largest group. - #[cfg_attr( - not(feature = "capture"), - allow(dead_code, reason = "only the `capture` feature uses it") - )] - track: moq_net::track::Producer, } impl Producer { @@ -141,7 +135,6 @@ impl Producer { reserved: moq_mux::catalog::Reserved, rendition: hang::catalog::VideoConfig, ) -> Result { - let published = track.clone(); let codecs = match &rendition.codec { hang::catalog::VideoCodec::H264(_) => Codecs::H264 { split: moq_mux::codec::h264::Split::new(), @@ -158,19 +151,22 @@ impl Producer { ))); } }; - Ok(Self { - codecs, - track: published, - }) + Ok(Self { codecs }) } /// A watch-only handle to the track's subscriber demand, created eagerly so /// subscription state is observable before any frames arrive. Watch it via /// [`wanted`](moq_net::track::Demand::wanted) / [`wanted_changed`](moq_net::track::Demand::wanted_changed). pub fn demand(&self) -> moq_net::track::Demand { + self.track().demand() + } + + /// The published track. A frame that changes the decoder identity, such as a reopen onto + /// another mode or backend, moves the rendition to a newly named one. + fn track(&self) -> &moq_net::track::Producer { match &self.codecs { - Codecs::H264 { import, .. } => import.demand(), - Codecs::H265 { import, .. } => import.demand(), + Codecs::H264 { import, .. } => import.track(), + Codecs::H265 { import, .. } => import.track(), } } @@ -178,8 +174,9 @@ impl Producer { /// open with a keyframe, or `None` while the largest serves. #[cfg(feature = "capture")] fn wanted(&self) -> Option { - let wanted = self.track.demand().wanted()?; - let largest = self.track.largest().map(|largest| largest.group); + let track = self.track(); + let wanted = track.demand().wanted()?; + let largest = track.largest().map(|largest| largest.group); largest.is_none_or(|largest| wanted > largest).then_some(wanted) } @@ -497,12 +494,11 @@ impl Driver { }; let mut producer = Producer::with_track(track, catalog.reserve().with_hls_hint(gop), rendition)?; - let demand = producer.demand(); let result = tokio::select! { biased; () = controls.closed() => Ok(()), - result = capture_loop(&mut producer, &demand, &mut source, &options, &clock, &requests) => result, + result = capture_loop(&mut producer, &mut source, &options, &clock, &requests) => result, }; // This runs only when the loop ends on its own (the track is usually already @@ -709,7 +705,6 @@ async fn wait_capture( #[cfg(feature = "capture")] async fn capture_loop( producer: &mut Producer, - demand: &moq_net::track::Demand, source: &mut S, options: &Capture, clock: &moq_mux::Clock, @@ -721,6 +716,9 @@ async fn capture_loop( let mut reservation: Option = None; let counted = requests.consume(); let capture = &options.capture; + // The published track's demand, followed onto the next track when a frame moves the rendition. + let mut demand = producer.demand(); + let demand = &mut demand; loop { // Idle until a viewer subscribes; the track ending is a clean exit. The @@ -769,16 +767,18 @@ async fn capture_loop( // larger mode below what it can send, or keep claiming room a smaller one no // longer needs. let ceiling = encoder_config.resolved_bitrate(); - let reservation = reservation.get_or_insert_with(|| options.bandwidth.reserve(demand, ceiling)); - reservation.update(ceiling); + let claim = reservation.get_or_insert_with(|| options.bandwidth.reserve(demand, ceiling)); + claim.update(ceiling); // Rate control is per encoder: this one opened at the configured bitrate, // so the policy's ceiling is that rate and the target starts there. A // reopened camera starts optimistic again rather than inheriting the // backed-off rate from whatever the link was doing last time. - let mut rate = Some(rate::Follower::new(reservation.consume(), rate::Policy::new(ceiling))); + let mut rate = Some(rate::Follower::new(claim.consume(), rate::Policy::new(ceiling))); // Per encoder, since a fresh one opens with a keyframe that serves every earlier ask. let mut forced = Cuts::new(served); + // Set when a frame moves the rendition to a newly named track. + let mut moved = false; loop { // Race the next frame against the last viewer leaving so we release the @@ -817,6 +817,13 @@ async fn capture_loop( break; }; producer.publish(&encoded)?; + // A frame whose decoder identity differs from the probed one (a reopen onto another + // mode or backend) publishes under a new name and ends this track. Its viewers find + // the new name in the catalog, so idle until one subscribes there. + if producer.track().name() != demand.name() { + moved = true; + break; + } } // Drop the camera (LED off) and encoder before waiting for the next viewer. @@ -824,6 +831,12 @@ async fn capture_loop( drop(encoder); capture_stopped(producer)?; tracing::info!("capture stopped; released source"); + if moved { + *demand = producer.demand(); + tracing::info!(track = %demand.name(), "capture moved to a new track"); + // The claim watches the old track's demand, so take a new one against the new track. + reservation = None; + } } } @@ -975,7 +988,7 @@ mod tests { #[tokio::test] #[cfg(feature = "openh264")] - async fn source_resize_updates_the_published_rendition() { + async fn source_resize_publishes_a_new_track() { let mut broadcast = moq_net::broadcast::Info::new().produce(); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); let mut initial = Config::new(320, 240, crate::FrameRate::new(30, 1).unwrap()); @@ -1002,7 +1015,9 @@ mod tests { capture_stopped(&mut producer).unwrap(); } - let (_, rendition) = rendition(&catalog).expect("the resized rendition should be published"); + // 640x360 takes a higher level than the probed 320x240, so it is a new track. + let (name, rendition) = rendition(&catalog).expect("the resized rendition should be published"); + assert!(name.ends_with(".avc3/1"), "{name} is not a new track"); assert_eq!(rendition.coded_width, Some(640)); assert_eq!(rendition.coded_height, Some(360)); } @@ -1211,7 +1226,6 @@ mod tests { probe.probe().await.unwrap(), ) .unwrap(); - let demand = producer.demand(); let (opens, rx) = tokio::sync::mpsc::unbounded_channel(); let (stop, stopped) = tokio::sync::oneshot::channel::<()>(); @@ -1228,7 +1242,6 @@ mod tests { tokio::select! { res = capture_loop( &mut producer, - &demand, &mut source, &options, &clock, @@ -1268,8 +1281,13 @@ mod tests { /// Plug in the camera the loop opens next. fn camera(&self) -> Synthetic { + self.camera_sized(320, 240) + } + + /// Plug in a camera of another mode for the loop to open next. + fn camera_sized(&self, width: u32, height: u32) -> Synthetic { let (camera, stream) = - Synthetic::open(crate::Size::new(320, 240), crate::FrameRate::new(30, 1).unwrap()); + Synthetic::open(crate::Size::new(width, height), crate::FrameRate::new(30, 1).unwrap()); self.opens.send(stream).unwrap(); camera } @@ -1298,10 +1316,14 @@ mod tests { } pub(super) fn surface() -> crate::frame::Surface { + surface_sized(320, 240) + } + + fn surface_sized(width: u32, height: u32) -> crate::frame::Surface { crate::frame::Surface::I420(crate::frame::I420 { - width: 320, - height: 240, - data: vec![0x80; 320 * 240 * 3 / 2], + width, + height, + data: vec![0x80; usize::try_from(width * height * 3 / 2).unwrap()], color: None, }) } @@ -1433,6 +1455,45 @@ mod tests { .await } + /// A reopen onto a mode the probed level cannot carry is a new codec string, so it moves + /// the rendition to a new track: the old one ends, and a viewer who follows the catalog + /// there reopens the camera on it. + #[tokio::test] + async fn a_reopen_onto_a_new_level_follows_the_new_track() { + tokio::task::LocalSet::new() + .run_until(async { + let fixture = Fixture::start(Duration::from_secs(1), moq_clock::wall()).await; + let probed = fixture.catalog.snapshot().video.renditions.keys().next().unwrap().clone(); + let mut track = fixture.subscribe().await; + let camera = fixture.camera(); + camera.push_native(surface(), us(0)); + read(&mut track).await; + + // The device goes away and comes back in a mode four times the size. + camera.close(); + let camera = fixture.camera_sized(640, 480); + camera.push_native(surface_sized(640, 480), us(40_000)); + // The probed track ends without the reopened camera's frames. + while let Some(read) = track.read().await.unwrap() { + assert!( + !matches!(read, moq_mux::container::Read::Frame { .. }), + "the reopened camera publishes on the probed track" + ); + } + + let mut moved = fixture.subscribe().await; + let name = fixture.catalog.snapshot().video.renditions.keys().next().unwrap().clone(); + assert_eq!(name, format!("{probed}/1")); + let camera = fixture.camera_sized(640, 480); + camera.push_native(surface_sized(640, 480), us(80_000)); + let first = read(&mut moved).await; + let second = read_new(&mut moved, &[first]).await; + assert!(second > first, "the new track carries the reopened camera"); + fixture.finish().await; + }) + .await + } + /// The wall mapping is pinned when the broadcast clock is built. A system clock stepped /// an hour since then retimes neither the published timestamps nor the advertised mapping. #[tokio::test] @@ -1855,7 +1916,6 @@ mod tests { rendition.probe().await.unwrap(), ) .unwrap(); - let demand = producer.demand(); let (camera, stream) = camera(); let mut source = Streams::new([stream]); @@ -1870,7 +1930,6 @@ mod tests { let clock = catalog.clock(); let mut run = std::pin::pin!(capture_loop( &mut producer, - &demand, &mut source, &options, &clock, From f1ea7443155fee1ed940018f3e5e492b34fe9e66 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 09:15:08 -0700 Subject: [PATCH 08/14] docs: describe Rust identity enforcement and finish its quest Co-Authored-By: Claude Opus 5.5 --- doc/concept/hang.md | 8 ++-- doc/lib/rs/moq-mux.md | 4 ++ quest/m1/README.md | 1 - quest/m1/archive/README.md | 1 - quest/m1/catalog-identity-rs.md | 82 --------------------------------- quest/m2/publish-publication.md | 9 ++-- rs/moq-c/MIGRATE.md | 6 +++ rs/moq-mux/MIGRATE.md | 9 ++++ rs/moq-video/MIGRATE.md | 5 ++ 9 files changed, 31 insertions(+), 94 deletions(-) delete mode 100644 quest/m1/catalog-identity-rs.md diff --git a/doc/concept/hang.md b/doc/concept/hang.md index 41802ac4890..b8dab921033 100644 --- a/doc/concept/hang.md +++ b/doc/concept/hang.md @@ -50,13 +50,13 @@ every video rendition (display size, rotation, flip) sit at the section root. A track name always means the same content, so a catalog update never redefines a track. The fields a decoder can't change without reinitializing -are fixed: codec, description, container, display aspect, sample rate, and -channel count. Everything else, such as the coded size, `bitrate`, +are fixed: codec, description, container, display aspect, interlacing, sample +rate, and channel count. Everything else, such as the coded size, `bitrate`, `framerate`, and `enabled`, updates in place; the picture resizes in band at a keyframe unless a `description` pins it. A publisher that needs a new identity publishes a new track name, or restarts the broadcast under a new -[epoch](/concept/moq-lite#publisher-epochs). A player can configure its -decoder once per track. +[epoch](/concept/moq-lite#publisher-epochs); the Rust and `@moq/publish` +publishers take a new name. A player can configure its decoder once per track. A few things the catalog can express beyond decoder config: diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 9e7c3494d37..e6e682ac030 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -25,6 +25,10 @@ arrived. The catalog clock never moves once something uses it: an importer that starts after a data track, a capture, another importer, or a pinned clock shifts its timestamps onto that clock instead, landing its first frame at now. A `catalog::Timebase` gives renditions on one timestamp base one shared shift. A +stream whose decoder identity changes, such as an SPS with a new profile or level +or an audio header with a new sample rate, continues under a new track name +(`/` for the codec importers) while the old track ends; everything else, +such as the coded size, updates the rendition in place. A timeline that rewinds ends the import. fMP4 export fixes its track set at the init segment, so a new rendition or a changed configuration ends the export. The [CLI page](/bin/cli#import) covers these and the MPEG-TS specifics: diff --git a/quest/m1/README.md b/quest/m1/README.md index 3f93058e5c5..f2d4fc31ec2 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -154,7 +154,6 @@ removed). - [Mock reset codes](/quest/m1/mock-reset-codes.md) - `js/net`'s mock transport carries only reset codes, and any reset found sending the wrong code is fixed - [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 - [JS FETCH fill timeout](/quest/m1/ietf-fill-timeout-js.md) - the JS FETCH responder honors FILL_TIMEOUT as Rust does -- [Rust identity enforcement](/quest/m1/catalog-identity-rs.md) - `moq-mux` refuses a redefined rendition, and every Rust publisher mints a new rendition name for a decoder identity change and updates everything else in place - [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/archive/README.md b/quest/m1/archive/README.md index b62432a2fc2..6753a5d0edc 100644 --- a/quest/m1/archive/README.md +++ b/quest/m1/archive/README.md @@ -88,6 +88,5 @@ from a host. Each child fixes the stale docs it touches. ## Related -- [Rust identity enforcement](/quest/m1/catalog-identity-rs.md) - a track's definition never changes for its name, so the archived catalog describes every group of the tracks it lists - [e2ee](/quest/m2/e2ee/README.md) - protected broadcasts are excluded initially - [DVR rewind](/quest/m2/archive-dvr.md) - seeks back through the cache, then an archive, with time ranges diff --git a/quest/m1/catalog-identity-rs.md b/quest/m1/catalog-identity-rs.md deleted file mode 100644 index 6b0e6f99492..00000000000 --- a/quest/m1/catalog-identity-rs.md +++ /dev/null @@ -1,82 +0,0 @@ -# [L] Rust identity enforcement - -## Goal - -No Rust publisher redefines a track it has published. A change to a -rendition's identity, the fields a consumer cannot change without -reinitializing its decoder or demuxer (`codec`, `description`, `container`, -`display_aspect_width`/`display_aspect_height`, `interlaced`, `sample_rate`, -`channel_count`), publishes a new rendition name, per the hang draft's -[Track Identity](/drafts/draft-lcurley-moq-hang.md#identity) section. Every -other field, including coded size, `bitrate`, and `framerate`, updates in -place. A replayed archive catalog then describes every group of the tracks it -lists. - -## Plan - -- ✅ Decided 2026-10-13 (maintainer, #5459): split from the catalog identity - quest. The contract and docs landed there; `@moq/publish` enforces the - browser side. -- ✅ Decided 2026-10-13 (maintainer, #5459): the Opus description is probed - up front. -- ✅ Decided 2026-10-15 (maintainer): mint a new rendition name, not a new - epoch; reverses the #5459 decision. Restarting the whole broadcast is an - extreme response to one rendition changing, and a name always means the - same content, so changed content gets a new name. `@moq/publish` keeps the - catalog ID and points its `track` at the next free `/` (#5552); each - Rust violator mints a name likewise, as FLV, HLS import, and transcode rungs - already do. A new epoch stays reserved for a whole-broadcast restart. -- ✅ Decided 2026-10-15 (maintainer): only fields that force a decoder - reinitialization mint a new track; everything else updates in place. New - tracks have user impact, so they are minted only when they must be. This - drops the #5459 coded-size ceilings and the rule that `bitrate`/`framerate` - never fall: a decoder follows a resize in band, and `codec` names a level - covering every picture, so outgrowing the size takes a new name only when - the codec string changes. - -Guard `moq-mux` `Rendition::set`, `replace`, and `stage` -(`rs/moq-mux/src/catalog/tracks.rs`): refuse a change to identity, and let -every other field update. Then fix each violator the 2026-10-13 audit found, -minting a name where the identity changes and updating in place otherwise: - -- The h264/h265 importers (an inline SPS, or a new avcC/hvcC through - `initialize`), av1 (a new sequence header, including the av1C defaults and - the minimal placeholder), and vp8/vp9 (keyframe size; VP9's level follows - it) rewrite in place, including MPEG-TS and `moq-rtc` ingest through them. - A size change alone updates in place; a new description or codec string - mints a name. -- A codec hint published up front is republished with the real - profile/level by the first keyframe. Publish once from the stream, or mint - when the codec string differs. -- `moq-video` reopening onto another camera mode or backend - (`source_resize_updates_the_published_rendition` asserts the rewrite). -- `moq-c` re-declaring a config under the same name. -- TS AAC/MP2/AC-3 and FLV legacy audio fix the config from the first header - and silently ignore a later change; refuse or mint instead. A PMT that - changes a PID's stream type or descriptors already opens a new track. -- `moq-audio`/`moq-ffi` `set_bitrate` should update the catalog's bitrate. - -FLV, fMP4, MKV, HLS import, `moq-transcode` rungs, and the `moq-gst` sink -(since [moq-dev/moq#6780](https://github.com/moq-dev/moq/pull/6780), which -keeps the live producer for caps that build the same importer and mints -`/` otherwise) already comply; keep -their tests. Update the tests that assert in-place identity rewrites -(`tracks.rs`, `moq-video`, `moq-gst`, `moq-c`); a same-name resize, such as -the transcode fixture's, is now allowed. - -Later, at lower priority: watch and HLS export consumer coverage for in-band -resolution changes, rendition switches, reconnects, and late joiners against -the contract. - -Decided 2026-10-25 (maintainer), reversing the 2026-10-20 decision to wait -on the generated C package: the hand-written `moq-c` ships at this cut with -parity to moq-ffi for every m1 binding feature, so the moq-c fix lands there, -and the generated C line follows after the cut. - -The branch checkpoint `origin/quest/m1/catalog-identity-rs` predates the -2026-10-15 decision to mint a new rendition name, so rework it against that -decision rather than resuming it as is. - -## Related - -- [Broadcast archives](/quest/m1/archive/README.md) - an archived catalog describes every stored group by the identity contract diff --git a/quest/m2/publish-publication.md b/quest/m2/publish-publication.md index 9597f71cdd7..86ea4c9fd7c 100644 --- a/quest/m2/publish-publication.md +++ b/quest/m2/publish-publication.md @@ -17,11 +17,8 @@ d0337260a6). The state is spread over five name-keyed maps in `Broadcast` (`#kinds`, `#names`, `#replacements`, `#served`, `#listed`), plus a per-Broadcast `#advertised` WeakMap and three failure layers in each encoder. Replace them with one publication object per network broadcast, -mirroring Rust's identity rules from -[Rust identity enforcement](/quest/m1/catalog-identity-rs.md). +mirroring Rust's identity rules (`hang::catalog::VideoConfig::same_identity` +and `AudioConfig::same_identity`, enforced by `moq-mux` since +[moq-dev/moq#7110](https://github.com/moq-dev/moq/pull/7110)). Public API: none. Wire: none. - -## Required - -- [Rust identity enforcement](/quest/m1/catalog-identity-rs.md) - settles the identity contract this mirrors diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index 31737ddda7e..def360d2912 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Authored configs keep their identity ([#7110](https://github.com/moq-dev/moq/pull/7110)) + +`moq_publish_video_config` and `moq_publish_audio_config` return `MOQ_ERROR_MUX` for a repeat call that changes the rendition's `codec`, `description`, or `container`, or an audio `sample_rate` or `channel_count`. +Publish media with a new identity under a new name, and remove the old rendition. +Tracks published with `moq_publish_video` or `moq_publish_audio` take a new name such as `/1` when the stream's decoder identity changes, so read the name from the catalog rather than caching it. + ### Paths with an empty segment are refused ([#6648](https://github.com/moq-dev/moq/pull/6648)) Every function taking a path, such as `moq_origin_create_broadcast` and `moq_origin_request`, returns an error for a path with a leading, trailing, or doubled `/` instead of collapsing it. Pass the canonical path, such as `room/cam`. diff --git a/rs/moq-mux/MIGRATE.md b/rs/moq-mux/MIGRATE.md index 14b133c68a8..fff7604aaf8 100644 --- a/rs/moq-mux/MIGRATE.md +++ b/rs/moq-mux/MIGRATE.md @@ -4,6 +4,15 @@ How to update your code for each breaking change, newest first. ## Unreleased +### A decoder identity change publishes a new track ([#7110](https://github.com/moq-dev/moq/pull/7110)) + +A rendition never changes its identity in place: `codec`, `description`, `container`, `display_aspect_width`/`display_aspect_height`, and `interlaced` for video, and `codec`, `description`, `container`, `sample_rate`, and `channel_count` for audio (see `hang::catalog::VideoConfig::same_identity`). +`container::Producer::set`, `replace`, and `stage` refuse such a change with `Error::IdentityChanged`; every other field still updates in place. +The codec importers (H.264, H.265, AV1, VP8, VP9) publish a stream whose identity changes, such as a new SPS profile or level, or a new avcC, under the next free `/` and finish the old track. +TS AAC, MP2, AC-3, and E-AC-3, and FLV in-band audio start a new track when a frame header changes the format instead of ignoring it. +Read `name()`, `demand()`, and `track()` from the importer after each write rather than caching them. +An av1C without a sequence header in its `configOBUs` no longer publishes a rendition: the stream's first sequence header does. + ### `moq_net::Path` has no lifetime `Source::resolve_reference` takes `Option<&Relative>` and returns `Option`, `ts::Programs::new` takes `impl Into`, so pass a `Path`, and `follow::Instances::new` takes `impl TryInto>`. Replace `PathOwned` with `Path` and drop `to_owned()` and `into_owned()` on paths. See moq-net's migration for the rest. diff --git a/rs/moq-video/MIGRATE.md b/rs/moq-video/MIGRATE.md index 5ec8794dfec..694d0e87a1c 100644 --- a/rs/moq-video/MIGRATE.md +++ b/rs/moq-video/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### A capture reopened onto a new identity publishes a new track ([#7110](https://github.com/moq-dev/moq/pull/7110)) + +When a reopened camera or encoder emits a different codec string than the probed rendition, such as a higher level for a larger mode, `encode::Producer` publishes under the next free `/` and finishes the old track. +`Producer::demand` follows the current track, so read it again after `publish` rather than holding the first one. + ### The renderer refuses GPU frames it cannot import (#6806) `render::Renderer::render` no longer downloads a GPU surface it cannot import; it returns an error naming the cause and the fix. From 9df4cfeb75058a240ac9828dab8d4a923f013f1b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 09:19:11 -0700 Subject: [PATCH 09/14] style: cargo fmt Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/catalog/producer.rs | 12 ++++++++++-- rs/moq-mux/src/codec/av1/import.rs | 6 +++++- rs/moq-mux/src/container/ts/import.rs | 24 +++++++++++++++++++---- rs/moq-video/src/encode/producer.rs | 28 ++++++++++++++++++--------- 4 files changed, 54 insertions(+), 16 deletions(-) diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 75b521e59bb..c9e9237c120 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -801,7 +801,11 @@ impl Producer { /// handed out, for a rendition whose identity changed: a name always means the same content. /// /// A successor's own successor counts on from the same base, so `v/1` is followed by `v/2`. - pub(crate) fn fresh_track(&self, name: &str, info: moq_net::track::Info) -> crate::Result { + pub(crate) fn fresh_track( + &self, + name: &str, + info: moq_net::track::Info, + ) -> crate::Result { let mut state = take(&self.current); let base = match name.rsplit_once('/') { Some((base, n)) if n.parse::().is_ok() && state.revisions.contains_key(base) => base, @@ -810,7 +814,11 @@ impl Producer { let revision = state.revisions.entry(base.to_string()).or_default(); loop { *revision += 1; - match self.outputs.broadcast.create_track(format!("{base}/{revision}"), info.clone()) { + match self + .outputs + .broadcast + .create_track(format!("{base}/{revision}"), info.clone()) + { // Someone else published that name explicitly; try the next. Err(moq_net::Error::Duplicate) => continue, track => return Ok(track?), diff --git a/rs/moq-mux/src/codec/av1/import.rs b/rs/moq-mux/src/codec/av1/import.rs index 0d49e50a8d2..7aa93214c2a 100644 --- a/rs/moq-mux/src/codec/av1/import.rs +++ b/rs/moq-mux/src/codec/av1/import.rs @@ -405,7 +405,11 @@ mod tests { let mut frames = split.decode(&sequence, pts).unwrap(); frames.extend(split.flush(pts).unwrap()); import.decode(frames).unwrap(); - assert_eq!(import.track.name(), "video", "the repeated sequence header keeps the track"); + assert_eq!( + import.track.name(), + "video", + "the repeated sequence header keeps the track" + ); let color = catalog.snapshot().video.renditions["video"] .color .clone() diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 408a8ea3baa..917251bf4b5 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -3804,7 +3804,10 @@ impl AacStream { if let Some(import) = &mut self.import && format != self.format { - tracing::info!(name = import.name(), "AAC format changed mid-stream, starting a new track"); + tracing::info!( + name = import.name(), + "AAC format changed mid-stream, starting a new track" + ); import.finish()?; self.import = None; } @@ -4441,7 +4444,10 @@ impl LegacyStream { if let Some(import) = &mut self.import && format != self.format { - tracing::info!(name = import.name(), "audio format changed mid-stream, starting a new track"); + tracing::info!( + name = import.name(), + "audio format changed mid-stream, starting a new track" + ); import.finish()?; self.import = None; } @@ -6391,14 +6397,24 @@ pub(super) mod test { ..Default::default() }; let cc = mux.cc(AAC_PID); - mux.out.extend(audio_pes_packet(AAC_PID, cc, 90_000, &[frame(48_000), frame(48_000)].concat())); + mux.out.extend(audio_pes_packet( + AAC_PID, + cc, + 90_000, + &[frame(48_000), frame(48_000)].concat(), + )); import.decode(&mux.out).unwrap(); let first = catalog.snapshot().audio.renditions.keys().next().unwrap().clone(); mux.out.clear(); for pts in [180_000, 270_000] { let cc = mux.cc(AAC_PID); - mux.out.extend(audio_pes_packet(AAC_PID, cc, pts, &[frame(44_100), frame(44_100)].concat())); + mux.out.extend(audio_pes_packet( + AAC_PID, + cc, + pts, + &[frame(44_100), frame(44_100)].concat(), + )); } import.decode(&mux.out).unwrap(); let snapshot = catalog.snapshot(); diff --git a/rs/moq-video/src/encode/producer.rs b/rs/moq-video/src/encode/producer.rs index 35e6148ea08..82d344fcdf3 100644 --- a/rs/moq-video/src/encode/producer.rs +++ b/rs/moq-video/src/encode/producer.rs @@ -1463,7 +1463,15 @@ mod tests { tokio::task::LocalSet::new() .run_until(async { let fixture = Fixture::start(Duration::from_secs(1), moq_clock::wall()).await; - let probed = fixture.catalog.snapshot().video.renditions.keys().next().unwrap().clone(); + let probed = fixture + .catalog + .snapshot() + .video + .renditions + .keys() + .next() + .unwrap() + .clone(); let mut track = fixture.subscribe().await; let camera = fixture.camera(); camera.push_native(surface(), us(0)); @@ -1482,7 +1490,15 @@ mod tests { } let mut moved = fixture.subscribe().await; - let name = fixture.catalog.snapshot().video.renditions.keys().next().unwrap().clone(); + let name = fixture + .catalog + .snapshot() + .video + .renditions + .keys() + .next() + .unwrap() + .clone(); assert_eq!(name, format!("{probed}/1")); let camera = fixture.camera_sized(640, 480); camera.push_native(surface_sized(640, 480), us(80_000)); @@ -1928,13 +1944,7 @@ mod tests { }; let requests = control.requests.weak(); let clock = catalog.clock(); - let mut run = std::pin::pin!(capture_loop( - &mut producer, - &mut source, - &options, - &clock, - &requests, - )); + let mut run = std::pin::pin!(capture_loop(&mut producer, &mut source, &options, &clock, &requests,)); // Nothing is known before an encoder opens, and its opening keyframe serves the request. control.cut().unwrap(); From 696e0480896a1373f2d063804f0dfebe836db6da Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 09:32:02 -0700 Subject: [PATCH 10/14] fix(mux): a minted track keeps its predecessor's HLS segment duration Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/catalog/tracks.rs | 6 +++++- rs/moq-mux/src/container/producer.rs | 7 +++++-- rs/moq-mux/src/hls.rs | 5 +++++ 3 files changed, 15 insertions(+), 3 deletions(-) diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index acd6f4fd21a..46b526b8268 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -565,13 +565,17 @@ pub(crate) type VideoTrack = Rendition; pub(crate) type AudioTrack = Rendition; /// A single text (caption/subtitle) track's catalog rendition. See [`Rendition`]. impl> Rendition { + /// The successor of this rendition's track, under a newly minted name, for a decoder identity + /// change. `hls_hint` carries the segment duration the old track declared. pub(crate) fn fresh( &self, container: super::hang::Container, info: moq_net::track::Info, + hls_hint: Option, ) -> crate::Result> { let track = self.catalog.fresh_track(&self.name, info)?; - self.catalog.track(track, container, None) + let rendition = self.catalog.rendition(track.name())?; + self.catalog.media(track, container, rendition, None, hls_hint) } fn new(reserved: Reserved, name: String) -> crate::Result { diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index 04470fae928..2b0e7962dff 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -112,6 +112,7 @@ trait Rendition: Send { &self, container: crate::catalog::hang::Container, info: moq_net::track::Info, + hls_hint: Option, ) -> crate::Result>; } @@ -128,8 +129,9 @@ where &self, container: crate::catalog::hang::Container, info: moq_net::track::Info, + hls_hint: Option, ) -> crate::Result> { - self.fresh(container, info) + self.fresh(container, info, hls_hint) } fn timestamp(&self, hint: Option) -> crate::Result { @@ -280,7 +282,8 @@ impl Producer { return self.set(config); } let container = crate::catalog::hang::Container::try_from(&config)?; - let mut next = rendition.fresh(container, self.inner.info().clone())?; + let hls_hint = self.recorder.as_ref().and_then(|recorder| recorder.declared()); + let mut next = rendition.fresh(container, self.inner.info().clone(), hls_hint)?; next.buffer_duration = self.buffer_duration; next.set(config)?; self.finish()?; diff --git a/rs/moq-mux/src/hls.rs b/rs/moq-mux/src/hls.rs index 09c178688b0..0c91b9d707a 100644 --- a/rs/moq-mux/src/hls.rs +++ b/rs/moq-mux/src/hls.rs @@ -679,6 +679,11 @@ pub(crate) struct Recorder { } impl Recorder { + /// The track's fixed nominal segment duration, once known. + pub fn declared(&self) -> Option { + self.live.lock().unwrap().segmenter.declared() + } + /// Run `callback` once when an unhinted track learns its fixed duration. pub fn on_declare(&mut self, callback: impl FnOnce(Duration) + Send + 'static) { self.live.lock().unwrap().on_declare = Some(Box::new(callback)); From eb7abfdc80df650429d098dcb991e7d884e8aeaf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 09:53:51 -0700 Subject: [PATCH 11/14] fix(mux,ffi,c): retire a predecessor's HLS records and move its bandwidth claim A minted track's predecessor kept its HLS record track registered until the broadcast ended, and the moq-ffi and moq-c video producers kept reserving bandwidth against the ended track. Retire the predecessor once its recorder drops, on the codec importers and the TS audio paths, and re-reserve against the new track after the write that moves it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-c/MIGRATE.md | 1 + rs/moq-c/src/video.rs | 138 +++++++++++++++++++++----- rs/moq-ffi/src/video.rs | 97 ++++++++++++++++-- rs/moq-mux/src/catalog/timebase.rs | 6 ++ rs/moq-mux/src/catalog/tracks.rs | 5 + rs/moq-mux/src/codec/video.rs | 2 + rs/moq-mux/src/container/producer.rs | 16 ++- rs/moq-mux/src/container/ts/import.rs | 10 ++ 8 files changed, 239 insertions(+), 36 deletions(-) diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index def360d2912..ad146032072 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -9,6 +9,7 @@ How to update your code for each breaking change, newest first. `moq_publish_video_config` and `moq_publish_audio_config` return `MOQ_ERROR_MUX` for a repeat call that changes the rendition's `codec`, `description`, or `container`, or an audio `sample_rate` or `channel_count`. Publish media with a new identity under a new name, and remove the old rendition. Tracks published with `moq_publish_video` or `moq_publish_audio` take a new name such as `/1` when the stream's decoder identity changes, so read the name from the catalog rather than caching it. +A `moq_encode_video` frame that moves the rendition to a new track, such as a resume at a ceiling past the declared level, reserves bandwidth against the new track, so call `moq_encode_video_reservation` again rather than holding the first handle. ### Paths with an empty segment are refused ([#6648](https://github.com/moq-dev/moq/pull/6648)) diff --git a/rs/moq-c/src/video.rs b/rs/moq-c/src/video.rs index 248a2dd3493..f9d3d231a6d 100644 --- a/rs/moq-c/src/video.rs +++ b/rs/moq-c/src/video.rs @@ -390,6 +390,9 @@ pub(crate) struct VideoEncoder { /// The encoded resolution, from the publish config. Frames carry only pixels, /// so this is what says how to read them. size: moq_video::Size, + /// The allocator [`follow`](VideoPublish::follow) reserved against, kept to reserve again + /// when a decoder identity change moves the rendition to a new track. + allocator: Option, reservation: Option>, follow: Option>, followed: Option, @@ -499,8 +502,41 @@ fn finalize( } } +/// Reserve the encoder's ceiling against its current track and follow the grant, replacing any +/// earlier claim and its follower: a decoder identity change moves the rendition to a new track, +/// and the claim has to watch that track's demand. +fn follow_grant(shared: &Shared) -> Result<(), Error> { + let (reservation, ceiling, lower) = { + let mut guard = shared.lock(); + let encoder = guard.as_mut().ok_or(Error::MediaNotFound)?; + let Some(allocator) = &encoder.allocator else { + return Ok(()); + }; + let ceiling = *encoder.ceiling.read(); + let reservation = Arc::new(allocator.reserve(&encoder.producer.demand(), ceiling)); + encoder.reservation = Some(reservation.clone()); + // Dropping the sender stops the previous follower. + encoder.follow = None; + (reservation, ceiling, encoder.lower.clone()) + }; + let (close, closed) = oneshot::channel(); + shared.lock().as_mut().ok_or(Error::MediaNotFound)?.follow = Some(close); + let follower = shared.clone(); + let background = crate::ffi::Background::new(async move { + tokio::select! { + biased; + _ = closed => {} + _ = follow_reservation(follower, lower, reservation.consume(), ceiling) => {} + } + Ok(()) + }); + shared.lock().as_mut().ok_or(Error::MediaNotFound)?.followed = Some(background); + Ok(()) +} + impl VideoEncoder { - fn publish_frame(&mut self, timestamp_us: u64, data: &[u8]) -> Result<(), Error> { + /// Encode and publish one picture, returning whether it moved the rendition to a new track. + fn publish_frame(&mut self, timestamp_us: u64, data: &[u8]) -> Result { if let Some(followed) = &self.followed { followed.check()?; } @@ -516,14 +552,17 @@ impl VideoEncoder { // Disabled: nothing is encoded until the grant resumes the rendition. let Some(encoder) = self.encoder.as_mut() else { - return Ok(()); + return Ok(false); }; let frame = moq_video::Frame::new(surface, moq_net::Timestamp::from_micros(timestamp_us)?); // A backend that pipelines hands back an earlier frame's output, so this is // zero or more access units rather than one per call. let encoded = block_on(encoder.encode(frame))?; + let before = self.producer.demand(); self.producer.publish(&encoded)?; - Ok(()) + // A frame whose decoder identity changed, such as a resume at a ceiling past the declared + // level, moved the rendition to a newly named track. + Ok(self.producer.demand().name() != before.name()) } fn publish_cut(&mut self) -> Result<(), Error> { @@ -688,6 +727,7 @@ impl Video { lower, format, size: config.size(), + allocator: None, reservation: None, follow: None, followed: None, @@ -702,27 +742,8 @@ impl Video { /// Reserve the encoder's ceiling against `allocator` and follow the grant. pub(crate) fn follow(&self, id: Id, allocator: &moq_net::bandwidth::Allocator) -> Result<(), Error> { let shared = self.producer(id)?; - let (reservation, ceiling, lower) = { - let mut guard = shared.lock(); - let encoder = guard.as_mut().ok_or(Error::MediaNotFound)?; - let ceiling = *encoder.ceiling.read(); - let reservation = Arc::new(allocator.reserve(&encoder.producer.demand(), ceiling)); - encoder.reservation = Some(reservation.clone()); - (reservation, ceiling, encoder.lower.clone()) - }; - let (close, closed) = oneshot::channel(); - shared.lock().as_mut().ok_or(Error::MediaNotFound)?.follow = Some(close); - let follower = shared.clone(); - let background = crate::ffi::Background::new(async move { - tokio::select! { - biased; - _ = closed => {} - _ = follow_reservation(follower, lower, reservation.consume(), ceiling) => {} - } - Ok(()) - }); - shared.lock().as_mut().ok_or(Error::MediaNotFound)?.followed = Some(background); - Ok(()) + shared.lock().as_mut().ok_or(Error::MediaNotFound)?.allocator = Some(allocator.clone()); + follow_grant(&shared) } pub(crate) fn reservation(&self, id: Id) -> Result>, Error> { @@ -1038,11 +1059,16 @@ pub unsafe extern "C" fn moq_encode_video_frame(producer: u32, frame: *const moq let data = unsafe { ffi::parse_slice(frame.data, frame.data_size)? }; let producer = State::lock().video.producer(producer)?; - producer + let moved = producer .lock() .as_mut() .ok_or(Error::MediaNotFound)? - .publish_frame(frame.timestamp_us, data) + .publish_frame(frame.timestamp_us, data)?; + // The follower takes the encoder's lock, so it moves with the track after the encode. + match moved { + true => crate::video::follow_grant(&producer), + false => Ok(()), + } }) } @@ -1289,6 +1315,65 @@ mod tests { (producer, track) } + /// A resume that reopens the encoder onto a larger picture publishes a new track (a higher + /// level is a new codec string), and the bandwidth claim moves there with it rather than + /// watching the track that ended. + #[tokio::test] + async fn a_resume_onto_a_new_level_reserves_on_the_new_track() { + let (producer, _track) = track_under_test().await; + let first = producer.demand().name().to_string(); + let probed = moq_video::encode::Config::new(320, 240, moq_video::FrameRate::new(30, 1).unwrap()); + let mut config = moq_video::encode::Config::new(640, 480, moq_video::FrameRate::new(30, 1).unwrap()); + config.kind = moq_video::encode::Kind::Software; + let mut opened = probed.clone(); + opened.kind = moq_video::encode::Kind::Software; + let estimate = moq_net::bandwidth::Producer::new(); + let shared = Shared::new(VideoEncoder { + encoder: Some(moq_video::encode::Sink::open(&opened).await.unwrap()), + producer, + config: config.clone(), + priority: hang::catalog::PRIORITY.video, + lower: None, + format: moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_RGBA, + size: probed.size(), + allocator: Some(moq_net::bandwidth::Allocator::new(estimate.consume())), + reservation: None, + follow: None, + followed: None, + ceiling: kio::Producer::new(moq_net::bandwidth::Rate::from_mbps(1)), + children: Default::default(), + }); + let moved = tokio::task::spawn_blocking(move || { + follow_grant(&shared).unwrap(); + let before = shared.lock().as_ref().unwrap().reservation.clone().unwrap(); + let small = vec![0x80; 320 * 240 * 4]; + assert!(!shared.lock().as_mut().unwrap().publish_frame(0, &small).unwrap()); + + // Disabled, then resumed onto the larger picture. + { + let mut guard = shared.lock(); + let video = guard.as_mut().unwrap(); + video.apply(moq_mux::rate::Target::Disabled).unwrap(); + video.size = config.size(); + video + .apply(moq_mux::rate::Target::Enabled(moq_net::bandwidth::Rate::from_mbps(1))) + .unwrap(); + } + let large = vec![0x80; 640 * 480 * 4]; + let moved = shared.lock().as_mut().unwrap().publish_frame(66_666, &large).unwrap(); + assert!(moved, "the larger picture is a new codec string"); + follow_grant(&shared).unwrap(); + let guard = shared.lock(); + let video = guard.as_ref().unwrap(); + let after = video.reservation.clone().unwrap(); + assert!(!Arc::ptr_eq(&before, &after), "the claim still watches the ended track"); + video.producer.demand().name().to_string() + }) + .await + .unwrap(); + assert_eq!(moved, format!("{first}/1")); + } + /// A clean finish reaches the subscriber as the end of the track, which is what /// makes the abort case below meaningful rather than vacuous. #[tokio::test] @@ -1346,6 +1431,7 @@ mod tests { lower: None, format: moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_I420, size: config.size(), + allocator: None, reservation: None, follow: None, followed: Some(followed), diff --git a/rs/moq-ffi/src/video.rs b/rs/moq-ffi/src/video.rs index f837d2b9d4f..459d29c2619 100644 --- a/rs/moq-ffi/src/video.rs +++ b/rs/moq-ffi/src/video.rs @@ -506,11 +506,43 @@ pub struct MoqVideoProducer { follow: std::sync::Mutex>, /// Last bitrate applied by the follow loop or [`set_bitrate`](Self::set_bitrate). applied: Arc, - /// Held so the reservation's registry outlives extra bandwidth handles. - _bandwidth: Option>, + /// The next lower rendition's ceiling, kept to follow a new track with. + lower: Option>, + /// Held so the reservation's registry outlives extra bandwidth handles, and to reserve + /// against the track a decoder identity change moves the rendition to. + bandwidth: Option>, children: crate::producer::Children, } +impl MoqVideoProducer { + /// Move the bandwidth claim and its follower onto `demand`, the track a decoder identity change + /// moved the rendition to: the old track's demand ended with it. + /// + /// The follower takes the producer's lock, so call this without holding it. + fn follow_track(&self, demand: &moq_net::track::Demand, ceiling: moq_net::bandwidth::Rate) -> Result<(), MoqError> { + let Some(bandwidth) = &self.bandwidth else { + return Ok(()); + }; + let stopped = self + .follow + .lock() + .unwrap() + .take() + .map_or(Ok(()), |mut follow| follow.stop()); + let reservation = bandwidth.reserve_demand(demand, ceiling.as_bps()); + let follow = spawn_follow(Followed { + inner: self.inner.clone(), + lower: self.lower.clone(), + consumer: reservation.consume(), + ceiling, + applied: self.applied.clone(), + }); + *self.reservation.lock().unwrap() = Some(reservation); + *self.follow.lock().unwrap() = Some(follow); + stopped + } +} + // Dropped or closed before its finish, the track ends the way a publisher leaving does. impl crate::producer::Child for MoqVideoProducer { fn abort(&self, err: moq_net::Error) { @@ -574,9 +606,20 @@ impl MoqVideoProducer { pub fn write(&self, frame: MoqVideoFrame) -> Result<(), MoqError> { self.check()?; let _guard = crate::ffi::runtime().enter(); - let mut guard = self.inner.lock().map_err(|_| panic_error())?; - let producer = guard.as_mut().ok_or(MoqError::Closed)?; - producer.write(frame) + let moved = { + let mut guard = self.inner.lock().map_err(|_| panic_error())?; + let producer = guard.as_mut().ok_or(MoqError::Closed)?; + let before = producer.producer.demand(); + producer.write(frame)?; + // A frame whose decoder identity changed, such as a resume at a ceiling past the + // declared level, moved the rendition to a newly named track. + let after = producer.producer.demand(); + (after.name() != before.name()).then(|| (after, *producer.ceiling.read())) + }; + match moved { + Some((demand, ceiling)) => self.follow_track(&demand, ceiling), + None => Ok(()), + } } /// Cut a new group at the next written frame. @@ -649,7 +692,9 @@ impl MoqVideoProducer { /// This encoder's bandwidth reservation, if it was published against a /// [`MoqAllocator`]. Dropping the handle does not release the claim; the - /// producer holds it until [`finish`](Self::finish). + /// producer holds it until [`finish`](Self::finish). A frame that changes the + /// decoder identity publishes under a new track name and reserves against that + /// track instead, so read this again after [`write`](Self::write). pub fn reservation(&self) -> Option> { self.reservation.lock().unwrap().clone() } @@ -801,7 +846,7 @@ impl MoqVideoProducer { let follow = reserved.as_ref().map(|reservation| { spawn_follow(Followed { inner: inner.clone(), - lower, + lower: lower.clone(), consumer: reservation.consume(), ceiling, applied: applied.clone(), @@ -813,7 +858,8 @@ impl MoqVideoProducer { reservation: std::sync::Mutex::new(reserved), follow: std::sync::Mutex::new(follow), applied, - _bandwidth: bandwidth, + lower, + bandwidth, children: children.clone(), })) } @@ -1763,6 +1809,41 @@ mod rung_tests { assert_eq!(lower_ceiling(&lower, &waiter), None); } + /// A rung resumed at a ceiling past its declared level publishes a new track, and its + /// bandwidth claim moves there with it rather than watching the track that ended. + #[tokio::test] + async fn a_rung_resumed_past_its_level_reserves_on_the_new_track() { + let ladder = ladder(); + ladder.high.write(frame(0)).unwrap(); + let before = ladder.high.reservation().unwrap(); + let apply = |target| { + let mut guard = ladder.high.inner.lock().unwrap(); + guard.as_mut().unwrap().apply(target).unwrap(); + }; + apply(Target::Disabled); + // 3 Mbps takes H.264 level 2.1, past the 2.0 the 2 Mbps rung declared. + ladder.high.set_bitrate(3_000_000).unwrap(); + apply(Target::Enabled(Rate::from_kbps(3_000))); + ladder.high.write(frame(1)).unwrap(); + + let names = ladder + .broadcast + .with_state(|state| { + Ok(state + .catalog + .snapshot() + .video + .renditions + .into_keys() + .collect::>()) + }) + .unwrap(); + assert!(names.contains(&"high/1".to_string()), "{names:?} has no new track"); + let after = ladder.high.reservation().unwrap(); + assert!(!Arc::ptr_eq(&before, &after), "the claim still watches the ended track"); + ladder.high.finish().unwrap(); + } + /// A disabled rung advertises itself disabled, drops the frames written to it, and /// resumes on a keyframe in a new group, its advertised maximum untouched throughout. #[tokio::test] diff --git a/rs/moq-mux/src/catalog/timebase.rs b/rs/moq-mux/src/catalog/timebase.rs index 36640f52750..0b74b2d3f71 100644 --- a/rs/moq-mux/src/catalog/timebase.rs +++ b/rs/moq-mux/src/catalog/timebase.rs @@ -29,6 +29,12 @@ impl Timebase { } } + /// Finish and forget the HLS records of a dropped `track` whose name is never published again. + /// See [`Producer::retire`]. + pub(crate) fn retire(&self, track: &str) -> crate::Result<()> { + self.catalog.retire(track) + } + /// Begin reserving an importer's tracks, sharing this timebase's offset. /// /// See [`Producer::reserve`] for how the reservation gates the catalog. diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index 46b526b8268..f033ddb87ca 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -578,6 +578,11 @@ impl> Rendition { self.catalog.media(track, container, rendition, None, hls_hint) } + /// Finish and forget the HLS records of `track`, a predecessor this rendition replaced. + pub(crate) fn retire(&self, track: &str) -> crate::Result<()> { + self.catalog.retire(track) + } + fn new(reserved: Reserved, name: String) -> crate::Result { Self::owned(reserved.producer(), Some(reserved), name) } diff --git a/rs/moq-mux/src/codec/video.rs b/rs/moq-mux/src/codec/video.rs index 28e5a15cd6a..3f29c11d21b 100644 --- a/rs/moq-mux/src/codec/video.rs +++ b/rs/moq-mux/src/codec/video.rs @@ -330,6 +330,7 @@ mod tests { let mut config = hang::catalog::VideoConfig::new("avc1.42c01e".parse::().unwrap()); config.description = Some(bytes::Bytes::from_static(b"first")); publishing.publish(&mut track, config.clone()).unwrap(); + let records = catalog.hls_tracks(); track .write(Frame { timestamp: moq_net::Timestamp::ZERO, @@ -373,6 +374,7 @@ mod tests { publishing.publish(&mut track, config).unwrap(); assert_ne!(track.name(), "video"); assert_ne!(track.name(), second); + assert_eq!(catalog.hls_tracks(), records, "each predecessor's HLS records retire"); track.finish().unwrap(); } } diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index 2b0e7962dff..eb6e36eb073 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -114,6 +114,7 @@ trait Rendition: Send { info: moq_net::track::Info, hls_hint: Option, ) -> crate::Result>; + fn retire(&self, track: &str) -> crate::Result<()>; } impl Rendition for crate::catalog::tracks::Rendition @@ -134,6 +135,10 @@ where self.fresh(container, info, hls_hint) } + fn retire(&self, track: &str) -> crate::Result<()> { + self.retire(track) + } + fn timestamp(&self, hint: Option) -> crate::Result { self.timestamp(hint) } @@ -287,8 +292,15 @@ impl Producer { next.buffer_duration = self.buffer_duration; next.set(config)?; self.finish()?; - *self = next; - Ok(()) + // The old name is never published again, so its HLS records end with it. They can only + // retire once its recorder is gone. + let old = std::mem::replace(self, next); + let name = old.name().to_string(); + drop(old); + match &self.rendition { + Some(rendition) => rendition.retire(&name), + None => Ok(()), + } } } diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 917251bf4b5..b0e1ad37c4c 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -3809,7 +3809,10 @@ impl AacStream { "AAC format changed mid-stream, starting a new track" ); import.finish()?; + let name = import.name().to_string(); + // The name is never published again, so its HLS records end once its recorder is gone. self.import = None; + self.timebase.retire(&name)?; } let import = match &mut self.import { Some(import) => { @@ -4449,7 +4452,10 @@ impl LegacyStream { "audio format changed mid-stream, starting a new track" ); import.finish()?; + let name = import.name().to_string(); + // The name is never published again, so its HLS records end once its recorder is gone. self.import = None; + self.timebase.retire(&name)?; } let import = match &mut self.import { Some(import) => import, @@ -6363,6 +6369,7 @@ pub(super) mod test { let mono = [mp2_frame(0xAA), mp2_frame(0xBB)].concat(); import.decode(&audio_pes_packet(MP2_PID, 0, 90_000, &mono)).unwrap(); let first = catalog.snapshot().audio.renditions.keys().next().unwrap().clone(); + let records = catalog.hls_tracks(); let both = [stereo(0xCC), stereo(0xDD)].concat(); import.decode(&audio_pes_packet(MP2_PID, 1, 180_000, &both)).unwrap(); import.decode(&audio_pes_packet(MP2_PID, 2, 270_000, &both)).unwrap(); @@ -6372,6 +6379,7 @@ pub(super) mod test { let (name, config) = snapshot.audio.renditions.iter().next().unwrap(); assert_ne!(name, &first, "the stereo frames publish under a new name"); assert_eq!(config.channel_count, 2); + assert_eq!(catalog.hls_tracks(), records, "the mono track's HLS records retire"); import.finish().unwrap(); let frames = read_audio_frames(&consumer, &catalog).await; assert_eq!(frames[0].payload, stereo(0xCC), "the new track starts at the change"); @@ -6405,6 +6413,7 @@ pub(super) mod test { )); import.decode(&mux.out).unwrap(); let first = catalog.snapshot().audio.renditions.keys().next().unwrap().clone(); + let records = catalog.hls_tracks(); mux.out.clear(); for pts in [180_000, 270_000] { @@ -6422,6 +6431,7 @@ pub(super) mod test { let (name, config) = snapshot.audio.renditions.iter().next().unwrap(); assert_ne!(name, &first, "the 44.1 kHz frames publish under a new name"); assert_eq!(config.sample_rate, 44_100); + assert_eq!(catalog.hls_tracks(), records, "the 48 kHz track's HLS records retire"); import.finish().unwrap(); let frames = read_audio_frames(&consumer, &catalog).await; assert_eq!( From 0d47f01418239acd3f3f92ea2c9017f4b8a30428 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 10:00:30 -0700 Subject: [PATCH 12/14] chore(dart): regenerate bindings Co-Authored-By: Claude Opus 5.5 --- dart/moq_ffi/lib/src/moq.dart | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 7bc2bf2a57d..60d6d701e9e 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -21644,7 +21644,7 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqvideoproducer_finish() != 40208) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqvideoproducer_reservation() != 25410) { + if (uniffi_moq_ffi_checksum_method_moqvideoproducer_reservation() != 1052) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqvideoproducer_set_bitrate() != 19081) { From 204118f92f5054cc42cfcdd8c928b90594f32a44 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 10:10:29 -0700 Subject: [PATCH 13/14] fix(ffi): a media track producer's demand follows a minted track Co-Authored-By: Claude Opus 5.5 --- dart/moq_ffi/lib/src/moq.dart | 2 +- rs/moq-ffi/src/media.rs | 11 +++++------ rs/moq-ffi/src/test.rs | 35 +++++++++++++++++++++++++++++++++++ 3 files changed, 41 insertions(+), 7 deletions(-) diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 60d6d701e9e..b29c0ac384b 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -21291,7 +21291,7 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_cut() != 54122) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } - if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_demand() != 16488) { + if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_demand() != 51616) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_discontinuity() != diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 4bb3ce1f75a..c3b5397aba8 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -799,8 +799,6 @@ impl MoqMediaCatalogProducer { struct MediaProducer { // Boxed because the codec splitters/imports make this much larger than the container one. import: Box, - /// Subscriber demand (name/used/unused) for the one track this publishes. - demand: moq_net::track::Demand, } /// An importer for a container, which may publish several tracks. @@ -1004,13 +1002,11 @@ impl MoqMediaTrackStreamProducer { } impl MoqMediaTrackProducer { - /// Wrap a single-codec importer, capturing the demand handle its track exposes. + /// Wrap a single-codec importer. fn from_import(import: moq_mux::import::Track, children: &Children) -> Result, MoqError> { - let demand = import.demand(); children.adopt(Arc::new(Self { inner: std::sync::Mutex::new(Some(MediaProducer { import: Box::new(import), - demand, })), children: children.clone(), })) @@ -1020,10 +1016,13 @@ impl MoqMediaTrackProducer { #[uniffi::export] impl MoqMediaTrackProducer { /// A watch-only handle to whether this track has subscribers. + /// + /// A frame that changes the decoder identity publishes under a new track name, so read this + /// again after [`write_frame`](Self::write_frame) rather than holding the first handle. pub fn demand(&self) -> Result, MoqError> { let guard = self.inner.lock().unwrap(); Ok(MoqTrackDemand::new( - guard.as_ref().ok_or(MoqError::Closed)?.demand.clone(), + guard.as_ref().ok_or(MoqError::Closed)?.import.demand(), )) } diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 2234b94424f..079f218ac44 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -2746,6 +2746,41 @@ async fn video_publish_named_track() { assert!(catalog.video.contains_key("hd"), "catalog: {:?}", catalog.video.keys()); } +/// A keyframe whose SPS names another level is a new decoder identity, so the track producer moves +/// to a new track, and its demand follows rather than watching the one that ended. +#[tokio::test] +async fn video_track_demand_follows_a_new_level() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let hd = MoqMediaTrackProducer::video( + &broadcast, + MoqMediaTarget::Named { + name: Some("hd".into()), + }, + video_init(MoqVideoFormat::Avc3, h264_init()), + ) + .unwrap(); + let keyframe = |level: u8| { + let mut bytes = h264_init(); + // The SPS's level_idc, after its NAL header, profile, and constraint bytes. + bytes[7] = level; + bytes.extend_from_slice(&[0, 0, 0, 1, 0x65, 0x88, 0x84, 0x21]); + bytes + }; + hd.write_frame(MoqFrame { + payload: keyframe(0x1f), + timestamp_us: Some(0), + }) + .unwrap(); + assert_eq!(hd.demand().unwrap().name(), "hd"); + hd.write_frame(MoqFrame { + payload: keyframe(0x28), + timestamp_us: Some(40_000), + }) + .unwrap(); + assert_eq!(hd.demand().unwrap().name(), "hd/1"); + hd.finish().unwrap(); +} + /// The raw-stream producer frames Annex-B written in arbitrary chunks, and /// `finish` publishes the last access unit even with no start code after it. #[tokio::test] From 297dfded1b4b82ba19b9a2c828fbbc8564196899 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 10:22:41 -0700 Subject: [PATCH 14/14] fix(mux,audio): keep a retune's other fields, and move a successor's mpegts entry container::Producer::set_bitrate advertises a supplied bitrate without replacing the rest of the config, which moq-audio's set_bitrate now uses, restoring the encoder when the publish fails. A TS PID whose format change mints a new track moves its mpegts entry to the new name. Co-Authored-By: Claude Opus 5.5 --- rs/moq-audio/src/encode/producer.rs | 10 ++++- rs/moq-c/MIGRATE.md | 2 +- rs/moq-mux/src/catalog/tracks.rs | 12 ++++++ rs/moq-mux/src/container/producer.rs | 14 +++++++ rs/moq-mux/src/container/ts/import.rs | 53 +++++++++++++++++---------- 5 files changed, 68 insertions(+), 23 deletions(-) diff --git a/rs/moq-audio/src/encode/producer.rs b/rs/moq-audio/src/encode/producer.rs index f46c8065d72..2c979dc1c0c 100644 --- a/rs/moq-audio/src/encode/producer.rs +++ b/rs/moq-audio/src/encode/producer.rs @@ -295,10 +295,16 @@ impl Producer { /// Retune the live encoder to `bitrate`. pub fn set_bitrate(&mut self, bitrate: moq_net::bandwidth::Rate) -> Result<(), Error> { let encoder = self.encoder.as_mut().ok_or(moq_net::Error::Closed)?; + let previous = encoder.bitrate(); encoder.set_bitrate(bitrate)?; + // Advertise the configured target in place, so encoded-byte estimates cannot replace it. + if let Err(err) = self.track.set_bitrate(encoder.bitrate().as_bps()) { + // Keep the encoder on what the catalog still advertises. It accepted that rate before, + // so the publish error is the one to report. + let _ = encoder.set_bitrate(previous); + return Err(err.into()); + } self.bitrate = encoder.bitrate(); - // Pin the configured target so encoded-byte estimates cannot replace it. - self.track.set(encoder.catalog())?; Ok(()) } diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index ad146032072..2e871100814 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -9,7 +9,7 @@ How to update your code for each breaking change, newest first. `moq_publish_video_config` and `moq_publish_audio_config` return `MOQ_ERROR_MUX` for a repeat call that changes the rendition's `codec`, `description`, or `container`, or an audio `sample_rate` or `channel_count`. Publish media with a new identity under a new name, and remove the old rendition. Tracks published with `moq_publish_video` or `moq_publish_audio` take a new name such as `/1` when the stream's decoder identity changes, so read the name from the catalog rather than caching it. -A `moq_encode_video` frame that moves the rendition to a new track, such as a resume at a ceiling past the declared level, reserves bandwidth against the new track, so call `moq_encode_video_reservation` again rather than holding the first handle. +A frame passed to `moq_encode_video_frame` that moves the rendition to a new track, such as a resume at a ceiling past the declared level, reserves bandwidth against the new track, so call `moq_encode_video_reservation` again rather than holding the first handle. ### Paths with an empty segment are refused ([#6648](https://github.com/moq-dev/moq/pull/6648)) diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index f033ddb87ca..59594d3840d 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -750,6 +750,18 @@ impl> Rendition { self.stage(config)?.commit() } + /// Advertise `bitrate` as a supplied value that measurement no longer replaces, leaving every + /// other field, measured or not, as published. + pub(crate) fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()> { + let mut config = self.config()?; + let mut estimate = config.estimate(); + estimate.bitrate = Some(bitrate); + config.set_estimate(estimate); + self.replace(config)?; + self.supplied.bitrate = Some(bitrate); + Ok(()) + } + /// Write `config` into the catalog, leaving the caller to choose how the guard publishes it. fn stage(&mut self, config: C) -> crate::Result> { if !self.present { diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index eb6e36eb073..afc6802b9b8 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -108,6 +108,7 @@ trait Rendition: Send { fn config(&self) -> crate::Result; fn replace(&mut self, config: R) -> crate::Result<()>; fn estimate(&mut self, estimate: crate::catalog::Estimate) -> crate::Result<()>; + fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()>; fn fresh( &self, container: crate::catalog::hang::Container, @@ -158,6 +159,10 @@ where fn estimate(&mut self, estimate: crate::catalog::Estimate) -> crate::Result<()> { self.estimate(estimate) } + + fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()> { + self.supply_bitrate(bitrate) + } } impl Producer { @@ -232,6 +237,15 @@ where self.rendition.as_mut().ok_or(crate::Error::NotPublished)?.set(config) } + /// Advertise `bitrate` as the rendition's maximum, such as an encoder's retuned target, which + /// later measurement no longer replaces. Every other published field stays as it is. + pub fn set_bitrate(&mut self, bitrate: u64) -> crate::Result<()> { + self.rendition + .as_mut() + .ok_or(crate::Error::NotPublished)? + .supply_bitrate(bitrate) + } + /// Modify the published catalog config. /// /// Estimate fields the config left to detection at [`set`](Self::set) stay owned by detection: diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index b0e1ad37c4c..45bfaa00bea 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -199,9 +199,10 @@ pub struct Import { /// PMT ES-level descriptors per PID, stashed when a PMT is parsed so a decoded /// media track can record them (language, registration, ...) once its track exists. es_descriptors: HashMap>, - /// Decoded media PIDs already recorded into `mpegts.tracks`, so the reconcile in - /// [`Self::flush`] runs once per track rather than on every frame. - recorded_media: HashSet, + /// Decoded media PIDs already recorded into `mpegts.tracks`, by the track name recorded, so the + /// reconcile in [`Self::flush`] runs once per track rather than on every frame, and moves the + /// entry when a format change publishes the PID under a new name. + recorded_media: HashMap, /// The program-level descriptors the catalog advertises, rewritten in place as the PMT /// revises them. program_info: Vec, @@ -294,7 +295,7 @@ impl Import { sections: HashMap::new(), supports_mpegts, es_descriptors: HashMap::new(), - recorded_media: HashSet::new(), + recorded_media: HashMap::new(), program_info, listed: HashMap::new(), si, @@ -1125,8 +1126,8 @@ impl Import { units += self.finish_stream(pid)?; if let Some(stream) = self.streams.get(&pid) { // A verbatim entry leaves with its stream; a decoded track's is recorded here. - if self.recorded_media.remove(&pid) - && let Some(name) = stream.media_track_name() + if let Some(name) = self.recorded_media.remove(&pid) + && stream.media_track_name().is_some() && let Some(mpegts) = self.catalog.modify()?.ext.mpegts_mut() { mpegts.tracks.remove(&name); @@ -1654,27 +1655,29 @@ impl Import { /// once per track. No-op without the `mpegts` section, before the track exists, /// or for verbatim streams (which self-register). fn record_media_track(&mut self, pid: Pid) -> anyhow::Result<()> { - if !self.supports_mpegts || self.recorded_media.contains(&pid) { + if !self.supports_mpegts { return Ok(()); } - let (name, descriptors) = { - let Some(name) = self.streams.get(&pid).and_then(|s| s.media_track_name()) else { - return Ok(()); - }; - ( - name, - self.es_descriptors.get(&pid.as_u16()).cloned().unwrap_or_default(), - ) + let Some(name) = self.streams.get(&pid).and_then(|s| s.media_track_name()) else { + return Ok(()); }; + if self.recorded_media.get(&pid) == Some(&name) { + return Ok(()); + } + let descriptors = self.es_descriptors.get(&pid.as_u16()).cloned().unwrap_or_default(); if let Some(mpegts) = self.catalog.modify()?.ext.mpegts_mut() { + // A format change moved the PID to a new track, whose entry replaces the ended one's. + if let Some(previous) = self.recorded_media.get(&pid) { + mpegts.tracks.remove(previous); + } let entry = mpegts .tracks - .entry(name) + .entry(name.clone()) .or_insert_with(|| catalog::Track::new(pid.as_u16())); entry.pid = pid.as_u16(); entry.descriptors = descriptors; } - self.recorded_media.insert(pid); + self.recorded_media.insert(pid, name); Ok(()) } @@ -5966,9 +5969,9 @@ pub(super) mod test { } /// Read every retained frame of the single audio rendition in `catalog`. - async fn read_audio_frames( + async fn read_audio_frames( consumer: &moq_net::broadcast::Consumer, - catalog: &crate::catalog::Producer, + catalog: &crate::catalog::Producer, ) -> Vec { let name = catalog .snapshot() @@ -6398,7 +6401,11 @@ pub(super) mod test { let mut broadcast = moq_net::broadcast::Info::new().produce(); let consumer = broadcast.consume(); - let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); let mut import = super::Import::new(broadcast, catalog.reserve()); let mut mux = Mux { out: synth_pmt(&[(StreamType::AdtsAac, AAC_PID)], false), @@ -6432,6 +6439,12 @@ pub(super) mod test { assert_ne!(name, &first, "the 44.1 kHz frames publish under a new name"); assert_eq!(config.sample_rate, 44_100); assert_eq!(catalog.hls_tracks(), records, "the 48 kHz track's HLS records retire"); + let recorded: Vec<_> = snapshot.ext.mpegts.tracks.keys().cloned().collect(); + assert_eq!( + recorded, + vec![name.clone()], + "the PID's mpegts entry moves to the new track" + ); import.finish().unwrap(); let frames = read_audio_frames(&consumer, &catalog).await; assert_eq!(