diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 7bc2bf2a57d..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() != @@ -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) { 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 b1f871bd4e5..bf18a96778c 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -151,7 +151,6 @@ removed). - [SUBSCRIBE honors the properties opt-out](/quest/m1/subscribe-timestamps.md) - the Rust publisher sends untimed objects to a SUBSCRIBE that declines Track Properties, and the Rust subscriber asks for them when it needs timestamps - [JS FETCH timeout test](/quest/m1/js-fetch-timeout-test.md) - a JS joining FETCH still waiting on its SUBSCRIBE at the deadline is refused with TIMEOUT - [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 -- [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/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-audio/src/encode/producer.rs b/rs/moq-audio/src/encode/producer.rs index 6b067bc9a74..2c979dc1c0c 100644 --- a/rs/moq-audio/src/encode/producer.rs +++ b/rs/moq-audio/src/encode/producer.rs @@ -295,7 +295,15 @@ 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(); Ok(()) } @@ -1332,6 +1340,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() { diff --git a/rs/moq-c/MIGRATE.md b/rs/moq-c/MIGRATE.md index 31737ddda7e..2e871100814 100644 --- a/rs/moq-c/MIGRATE.md +++ b/rs/moq-c/MIGRATE.md @@ -4,6 +4,13 @@ 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. +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)) 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-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 77717339d49..7b336585d07 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -808,6 +808,74 @@ 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(), + preference: 0, + }; + 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); +} + +/// 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-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/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] diff --git a/rs/moq-ffi/src/video.rs b/rs/moq-ffi/src/video.rs index a50fe16113f..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] @@ -1804,7 +1885,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/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-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 5df43ad42b5..c9e9237c120 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -49,6 +49,10 @@ struct State { /// Holds estimate-driven publishes to one per [`ESTIMATE_WINDOW`]. estimates: Throttle, + + /// The last revision [`Producer::fresh_track`] handed out per base name, so a successor never + /// reuses a name. + revisions: std::collections::HashMap, } impl State { @@ -459,6 +463,7 @@ impl Producer { clock, anchored, estimates: Throttle::default(), + revisions: Default::default(), })), hls: config.hls, max_age: config.max_age, @@ -792,6 +797,35 @@ impl Producer { } } + /// Create the successor of track `name` under the next `/` this catalog has not + /// 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 { + 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, + _ => name, + }; + let revision = state.revisions.entry(base.to_string()).or_default(); + loop { + *revision += 1; + 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?), + } + } + } + /// Build the media [`container::Producer`](crate::container::Producer) for `track`, /// enrolling a video or audio track for HLS records unless [`Config::with_hls`] disabled them. pub(super) fn media( 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 018ef2a12eb..59594d3840d 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -367,6 +367,22 @@ impl RenditionConfig for hang::catalog::TextConfig { } } +/// 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)) + && !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)) + && !previous.same_identity(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 @@ -549,6 +565,24 @@ 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)?; + let rendition = self.catalog.rendition(track.name())?; + 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) } @@ -610,6 +644,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; @@ -715,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 { @@ -724,6 +771,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) @@ -781,6 +829,97 @@ 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 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); + 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(); diff --git a/rs/moq-mux/src/codec/av1/import.rs b/rs/moq-mux/src/codec/av1/import.rs index f5f568b384c..7aa93214c2a 100644 --- a/rs/moq-mux/src/codec/av1/import.rs +++ b/rs/moq-mux/src/codec/av1/import.rs @@ -3,7 +3,7 @@ //! Publishes raw AV1 (OBU-framed, inline sequence headers) on a single moq //! track and resolves the catalog rendition. The codec config comes from the //! sequence header the splitter packages into the first keyframe (scanned out of -//! the frame here), or from an av1C record handed to +//! the frame here), or from the one in an av1C record's `configOBUs` handed to //! [`initialize`](Import::initialize). A keyframe that can't be configured is an //! error; non-keyframes before the first config are written through to the //! producer, which reports [`MissingKeyframe`](crate::container::MissingKeyframe) @@ -62,7 +62,7 @@ impl Import { /// Resolve the codec config from a sequence header / av1C and other metadata. /// /// - **av1C** (leading `0x81` marker): the buffer is parsed as an - /// AV1CodecConfigurationRecord, which resolves the config. + /// AV1CodecConfigurationRecord, whose `configOBUs` sequence header resolves the config. /// - **raw OBUs**: any sequence header resolves the config. /// /// Optional, since the importer also self-initializes from the first keyframe. @@ -90,34 +90,20 @@ impl Import { Ok(()) } + /// Resolve the config from the sequence header in an av1C's `configOBUs`. + /// + /// The record's own header fields omit the colour the codec string carries, so a config built + /// from them alone would differ from the stream's first sequence header and force a new track. + /// A record without one publishes nothing: the first keyframe must carry it to be decodable. fn init_from_av1c(&mut self, data: &[u8]) -> crate::Result<()> { - let seq_profile = (data[1] >> 5) & 0x07; - let seq_level_idx = data[1] & 0x1F; - let tier = ((data[2] >> 7) & 0x01) == 1; - let high_bitdepth = ((data[2] >> 6) & 0x01) == 1; - let twelve_bit = ((data[2] >> 5) & 0x01) == 1; - - // Resolution is unknown from av1C; it's filled when the first sequence header arrives. - let mut config = hang::catalog::VideoConfig::new(hang::catalog::AV1 { - profile: seq_profile, - level: seq_level_idx, - tier: if tier { 'H' } else { 'M' }, - bitdepth: super::bitdepth(twelve_bit, high_bitdepth), - mono_chrome: ((data[2] >> 4) & 0x01) == 1, - chroma_subsampling_x: ((data[2] >> 3) & 0x01) == 1, - chroma_subsampling_y: ((data[2] >> 2) & 0x01) == 1, - chroma_sample_position: data[2] & 0x03, - color_primaries: 1, - transfer_characteristics: 1, - matrix_coefficients: 1, - full_range: false, - }); // configOBUs follow the four header bytes; samples need not repeat what they carry. A new // record starts a new coded sequence, whose HDR metadata is what its configOBUs carry. - config.color = color(&data[4..]); - let hdr = find_hdr(&Bytes::copy_from_slice(&data[4..])); + let obus = &data[4..]; + let hdr = find_hdr(&Bytes::copy_from_slice(obus)); self.catalog.hdr(&mut self.track, hdr, true)?; - self.apply_config(config)?; + if let Some(seq) = find_sequence_header(obus) { + self.configure_from_seq(&seq)?; + } Ok(()) } @@ -177,10 +163,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) } @@ -392,7 +375,8 @@ mod tests { } /// A track initialized from an av1C publishes the colour its configOBUs carry, since the - /// samples need not repeat it. + /// samples need not repeat it, and the first sync sample repeating its sequence header keeps + /// the track. #[tokio::test(start_paused = true)] async fn av1c_initialize_publishes_config_obu_color() { let mut broadcast = moq_net::broadcast::Info::new().produce(); @@ -402,7 +386,8 @@ mod tests { .unwrap(); let mut import = Import::new(track, catalog.reserve(), Default::default()).unwrap(); - let mut av1c = vec![0x81, 0x08, 0x0c, 0x00]; + let mut av1c = vec![0x81, 0x09, 0x4c, 0x00]; + av1c.extend(av1_config_obus()); av1c.extend(hdr10_obus()); import.initialize(&av1c).unwrap(); @@ -420,6 +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" + ); let color = catalog.snapshot().video.renditions["video"] .color .clone() @@ -430,8 +420,30 @@ mod tests { ); } + /// An av1C without a sequence header publishes nothing: its header fields omit the colour the + /// codec string carries, so the stream's first sequence header configures the track instead. + #[tokio::test(start_paused = true)] + async fn av1c_without_a_sequence_header_waits_for_the_stream() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); + let track = broadcast + .create_track("video", hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let mut import = Import::new(track, catalog.reserve(), Default::default()).unwrap(); + import.initialize(&[0x81, 0x09, 0x4c, 0x00]).unwrap(); + assert!(catalog.snapshot().video.renditions.is_empty()); + + let pts = moq_net::Timestamp::from_micros(0).unwrap(); + let mut split = super::super::Split::new(); + let mut frames = split.decode(&av1_sequence_header(), pts).unwrap(); + frames.extend(split.flush(pts).unwrap()); + import.decode(frames).unwrap(); + assert_eq!(import.track.name(), "video"); + assert_eq!(catalog.snapshot().video.renditions["video"].coded_width, Some(1920)); + } + /// Seeding with raw OBUs whose sequence header is new starts a new coded sequence, dropping the - /// HDR metadata the last one carried. + /// HDR metadata an earlier av1C carried. #[tokio::test(start_paused = true)] async fn a_new_raw_sequence_drops_stale_hdr_metadata() { let mut broadcast = moq_net::broadcast::Info::new().produce(); @@ -443,15 +455,14 @@ mod tests { let mut av1c = vec![0x81, 0x08, 0x0c, 0x00]; av1c.extend(hdr10_obus()); import.initialize(&av1c).unwrap(); - let color = || catalog.snapshot().video.renditions["video"].color.clone().unwrap(); - assert!(color().mastering.is_some()); import.initialize(&av1_sequence_header()).unwrap(); - assert_eq!((color().mastering, color().content_light), (None, None)); + let color = catalog.snapshot().video.renditions["video"].color.clone().unwrap(); + assert_eq!((color.mastering, color.content_light), (None, None)); } - /// A temporal unit of a temporal delimiter, the sequence header from `av1.mp4`'s av1C, and a frame. - fn av1_sequence_header() -> Vec { + /// The configOBUs of `av1.mp4`'s av1C: its sequence header. + fn av1_config_obus() -> Vec { let init = include_bytes!("../../container/fmp4/test_data/av1.mp4"); let mut cursor = std::io::Cursor::new(&init[..]); let moov = loop { @@ -463,8 +474,13 @@ mod tests { let mp4_atom::Codec::Av01(av01) = &moov.trak[0].mdia.minf.stbl.stsd.codecs[0] else { panic!("expected av01"); }; + av01.av1c.config_obus.clone() + } + + /// A temporal unit of a temporal delimiter, the sequence header from `av1.mp4`'s av1C, and a frame. + fn av1_sequence_header() -> Vec { let mut tu = vec![0x12, 0x00]; - tu.extend_from_slice(&av01.av1c.config_obus); + tu.extend(av1_config_obus()); tu.extend_from_slice(&[0x32, 0x01, 0x10]); tu } diff --git a/rs/moq-mux/src/codec/h264/import.rs b/rs/moq-mux/src/codec/h264/import.rs index 842f26a252f..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()?; @@ -195,10 +202,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) } @@ -441,7 +445,8 @@ mod tests { } /// A frame-only SPS declares its reorder depth, which holds for the track's life: a later SPS - /// declaring less or none keeps it, and one declaring more is refused. + /// declaring less or none keeps it, and one declaring more is refused. An SPS that changes the + /// decoder identity mints a new track, which declares its own. #[tokio::test(start_paused = true)] async fn a_declared_reorder_depth_holds_for_the_track() { use crate::codec::h264::fixtures::{PPS, SPS_FIELD, SPS_IPB, SPS_PYRAMID}; @@ -452,7 +457,7 @@ mod tests { let mut deep = Import::new(deep, catalog.reserve(), Default::default()).unwrap(); let mut shallow = Import::new(shallow, catalog.reserve(), Default::default()).unwrap(); let mut split = Split::new(); - let mut keyframe = |import: &mut Import, name: &str, ms: u64, sps: &[u8]| { + let mut keyframe = |import: &mut Import, ms: u64, sps: &[u8]| { let mut annexb = BytesMut::new(); for nal in [sps, PPS, idr] { annexb.extend_from_slice(&[0, 0, 0, 1]); @@ -463,18 +468,17 @@ mod tests { frames.extend(split.flush(pts).expect("flush keyframe")); import .decode(frames) - .map(|()| catalog.snapshot().video.renditions[name].reorder) + .map(|()| catalog.snapshot().video.renditions[import.name()].reorder) }; - assert_eq!(keyframe(&mut deep, "deep", 0, SPS_PYRAMID).unwrap(), Some(2)); - assert_eq!(keyframe(&mut deep, "deep", 40, SPS_IPB).unwrap(), Some(2)); - assert_eq!( - keyframe(&mut deep, "deep", 80, SPS_FIELD).unwrap(), - Some(2), - "fields declare none" - ); - assert_eq!(keyframe(&mut shallow, "shallow", 0, SPS_IPB).unwrap(), Some(1)); - assert!(keyframe(&mut shallow, "shallow", 40, SPS_PYRAMID).is_err()); + assert_eq!(keyframe(&mut deep, 0, SPS_PYRAMID).unwrap(), Some(2)); + assert_eq!(keyframe(&mut deep, 40, SPS_IPB).unwrap(), Some(2)); + assert_eq!(deep.name(), "deep"); + // A field-coded SPS is a new profile, level, and scan: a new track with its own bound. + assert_eq!(keyframe(&mut deep, 80, SPS_FIELD).unwrap(), None, "fields declare none"); + assert_eq!(deep.name(), "deep/1"); + assert_eq!(keyframe(&mut shallow, 0, SPS_IPB).unwrap(), Some(1)); + assert!(keyframe(&mut shallow, 40, SPS_PYRAMID).is_err()); } /// A new out-of-band avcC starts a new coded sequence, so in-band HDR metadata from the last one @@ -502,7 +506,10 @@ mod tests { let color = || catalog.snapshot().video.renditions["video"].color.clone().unwrap(); assert!(color().mastering.is_some()); + // The new avcC is a new description, so it publishes a new track. import.initialize(&avcc(SPS_BT709)).unwrap(); + assert_eq!(import.name(), "video/1"); + let color = || catalog.snapshot().video.renditions["video/1"].color.clone().unwrap(); assert_eq!(color().primaries, Some(1)); assert_eq!((color().mastering, color().content_light), (None, None)); } diff --git a/rs/moq-mux/src/codec/h265/import.rs b/rs/moq-mux/src/codec/h265/import.rs index 97c9a13066f..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()?; @@ -196,10 +203,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) } @@ -308,14 +313,16 @@ mod tests { let mut frames = split.decode(&au, pts).unwrap(); frames.extend(split.flush(pts).unwrap()); import.decode(frames).unwrap(); - let color = || catalog.snapshot().video.renditions["video"].color.clone(); - assert!(color().unwrap().mastering.is_some()); + let color = |name: &str| catalog.snapshot().video.renditions[name].color.clone(); + assert!(color("video").unwrap().mastering.is_some()); - // An SPS that signals no colour: nothing is left once the stale metadata goes. + // An SPS that signals no colour: nothing is left once the stale metadata goes. It is also a + // new profile and level, so it publishes a new track. import .initialize(&annexb(&[fixtures::VPS, fixtures::SPS, fixtures::PPS])) .unwrap(); - assert_eq!(color(), None); + assert_eq!(import.name(), "video/1"); + assert_eq!(color("video/1"), None); } /// An HDR10 keyframe publishes its VUI colour and the static HDR metadata its SEI carries. diff --git a/rs/moq-mux/src/codec/video.rs b/rs/moq-mux/src/codec/video.rs index 6d4708e6744..3f29c11d21b 100644 --- a/rs/moq-mux/src/codec/video.rs +++ b/rs/moq-mux/src/codec/video.rs @@ -245,18 +245,24 @@ 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. A decoder identity change finishes the previous track and mints a fresh name; + /// anything else updates the rendition in place. /// /// A declared `reorder` is fixed for the track's life, so a later sequence header declaring /// less or none keeps it, and one declaring more fails: frames reordered that deep would - /// break the bound consumers already clock the track by. + /// break the bound consumers already clock the track by. A minted track declares its own. pub(crate) fn publish(&mut self, track: &mut Track, mut config: hang::catalog::VideoConfig) -> crate::Result<()> { self.resolved = Some(config.clone()); self.hdr.apply(&mut config.color); self.hint.apply(&mut config); config.enabled = self.enabled; - if let Some(declared) = self.last.as_ref().and_then(|last| last.reorder) { + // A new decoder identity mints a new track, whose bound starts over. + let declared = self + .last + .as_ref() + .filter(|last| last.same_identity(&config)) + .and_then(|last| last.reorder); + if let Some(declared) = declared { match config.reorder { Some(reorder) if reorder > declared => { return Err(anyhow::Error::new(Deepened { declared, reorder }).into()); @@ -268,7 +274,7 @@ impl Catalog { return Ok(()); } tracing::debug!(name = ?track.name(), ?config, "starting track"); - track.set(config.clone())?; + track.reconfigure(config.clone())?; self.last = Some(config); Ok(()) } @@ -300,3 +306,75 @@ impl Catalog { } } } + +#[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(); + let records = catalog.hls_tracks(); + track + .write(Frame { + timestamp: moq_net::Timestamp::ZERO, + payload: bytes::Bytes::from_static(b"old"), + keyframe: true, + duration: None, + }) + .unwrap(); + let old = track.track().clone(); + 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.largest().map(|p| p.group), Some(0)); + assert!( + old.poll_closed(&kio::Waiter::noop()).is_ready(), + "the previous track ends" + ); + track + .write(Frame { + timestamp: moq_net::Timestamp::ZERO, + payload: bytes::Bytes::from_static(b"new"), + keyframe: true, + duration: None, + }) + .unwrap(); + assert_eq!(track.track().largest().map(|p| p.group), 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); + assert_eq!(catalog.hls_tracks(), records, "each predecessor's HLS records retire"); + track.finish().unwrap(); + } +} diff --git a/rs/moq-mux/src/codec/vp8/import.rs b/rs/moq-mux/src/codec/vp8/import.rs index 4d96d603979..fd1fc570396 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 e01ec13f99b..e851a6a9665 100644 --- a/rs/moq-mux/src/codec/vp9/import.rs +++ b/rs/moq-mux/src/codec/vp9/import.rs @@ -63,10 +63,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/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 08d482f472f..afc6802b9b8 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -108,6 +108,14 @@ 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, + info: moq_net::track::Info, + hls_hint: Option, + ) -> crate::Result>; + fn retire(&self, track: &str) -> crate::Result<()>; } impl Rendition for crate::catalog::tracks::Rendition @@ -119,6 +127,19 @@ where self.name() } + fn fresh( + &self, + container: crate::catalog::hang::Container, + info: moq_net::track::Info, + hls_hint: Option, + ) -> crate::Result> { + 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) } @@ -138,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 { @@ -212,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: @@ -254,6 +288,36 @@ 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 previous.same_identity(&config) { + return self.set(config); + } + let container = crate::catalog::hang::Container::try_from(&config)?; + 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()?; + // 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(()), + } + } +} + /// A published rendition config edited in place. pub struct Guard<'a, R> { rendition: &'a mut dyn Rendition, diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 4b48737537c..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, @@ -894,9 +895,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 +990,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, @@ -1122,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); @@ -1651,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(()) } @@ -3645,13 +3651,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 +3801,22 @@ 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()?; + 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) => { // The description already carries the layout, so a repeated program config @@ -3832,12 +3858,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 +4289,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 +4444,22 @@ 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()?; + 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, None => { @@ -4422,13 +4469,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) } }; @@ -5920,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() @@ -6300,6 +6349,111 @@ 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 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(); + + 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); + 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"); + } + + // 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().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), + ..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(); + let records = catalog.hls_tracks(); + + 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); + 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!( + 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-mux/src/error.rs b/rs/moq-mux/src/error.rs index aeaba4c0537..11f143a1098 100644 --- a/rs/moq-mux/src/error.rs +++ b/rs/moq-mux/src/error.rs @@ -305,6 +305,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), /// A rendition moved to a track whose media goes back before what the export already read for it. #[error( 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)); 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. diff --git a/rs/moq-video/src/encode/producer.rs b/rs/moq-video/src/encode/producer.rs index 3acba2f2f75..82d344fcdf3 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,61 @@ 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 +1932,6 @@ mod tests { rendition.probe().await.unwrap(), ) .unwrap(); - let demand = producer.demand(); let (camera, stream) = camera(); let mut source = Streams::new([stream]); @@ -1868,14 +1944,7 @@ mod tests { }; let requests = control.requests.weak(); let clock = catalog.clock(); - let mut run = std::pin::pin!(capture_loop( - &mut producer, - &demand, - &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();