From f20cc65435e6da62768bca3c4bfea2d0de324991 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 23:46:27 -0800 Subject: [PATCH 1/8] chore(quest): claim quest/m1/stream-ladder.md Quest-Claim: dbbdafc675edcf75 From ada086d29cb27ba63c3970595da3333f3de76ab3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:10:27 -0800 Subject: [PATCH 2/8] feat(ffi)!: byte-stream producers form a simulcast ladder Co-Authored-By: Claude Opus 5.5 --- rs/moq-ffi/src/media.rs | 60 ++++++++++-- rs/moq-ffi/src/test.rs | 94 +++++++++++++++++- rs/moq-mux/src/import/track.rs | 172 ++++++++++++++++++++++++++++++++- 3 files changed, 313 insertions(+), 13 deletions(-) diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index ef89eaa2d1f..057dbfe9c47 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -400,7 +400,37 @@ pub struct MoqVideoInit { /// it. This track then sits one priority below it, so a bandwidth allocator funds the /// lower rendition first. #[uniffi(default = None)] - pub lower: Option>, + pub lower: Option, +} + +/// A single-track media producer of either kind, such as a rendition's [`lower`](MoqVideoInit::lower). +#[derive(Clone, uniffi::Enum)] +pub enum MoqMediaProducer { + /// Imports complete encoded frames. + Track { + /// The frame producer. + producer: Arc, + }, + /// Imports a byte stream. + Stream { + /// The byte-stream producer. + producer: Arc, + }, +} + +impl MoqMediaProducer { + /// The priority of the video track this producer publishes. + fn video_priority(&self) -> Result { + match self { + Self::Track { producer } => { + let guard = producer.inner.lock().unwrap(); + guard.as_ref().ok_or(MoqError::Closed)?.video.ok_or_else(|| MoqError::Codec { + reason: "the lower rendition is not a video track".into(), + }) + } + Self::Stream { producer } => Ok(producer.inner.lock().unwrap().as_ref().ok_or(MoqError::Closed)?.priority), + } + } } /// Shares one source's timestamp offset across containers without keeping its broadcast open. @@ -571,12 +601,7 @@ impl MoqVideoInit { out.hint = hint.into(); } if let Some(lower) = self.lower { - let guard = lower.inner.lock().unwrap(); - let below = guard.as_ref().ok_or(MoqError::Closed)?; - let priority = below.video.ok_or_else(|| MoqError::Codec { - reason: "the lower rendition is not a video track".into(), - })?; - out.priority = priority.saturating_sub(1); + out.priority = lower.video_priority()?.saturating_sub(1); } Ok(out) } @@ -891,6 +916,8 @@ struct ContainerProducer { /// A byte-stream importer for one codec track, where frame boundaries are inferred. struct MediaStreamProducer { import: Box, + /// The video track's priority, which a higher rung takes one below. + priority: i8, } /// Imports a container into a broadcast, from chunks split anywhere such as a raw byte stream. @@ -1065,6 +1092,7 @@ impl MoqMediaTrackStreamProducer { let _guard = crate::ffi::enter(); let init = init.resolve()?; broadcast.with_state(|state| { + let priority = init.priority; let request = target.resolve(&state.broadcast, &init.format)?; let import = @@ -1077,6 +1105,7 @@ impl MoqMediaTrackStreamProducer { state.children.adopt(Arc::new(MoqMediaTrackStreamProducer { inner: std::sync::Mutex::new(Some(MediaStreamProducer { import: Box::new(import), + priority, })), children: state.children.clone(), })) @@ -1323,6 +1352,23 @@ impl MoqMediaTrackStreamProducer { Ok(()) } + /// Stop or resume publishing this rendition, advertised as its catalog `enabled`. + /// + /// Disable it when its sender stops for lack of bandwidth (a + /// [`MoqRateTarget::Disabled`](crate::bandwidth::MoqRateTarget::Disabled)): this marks a + /// discontinuity and drops every access unit written until it is enabled again, so the + /// stream may keep flowing. Enabled again, it resumes, and advertises itself enabled, at + /// the next keyframe in the stream. + pub fn set_enabled(&self, enabled: bool) -> Result<(), MoqError> { + let _guard = crate::ffi::enter(); + let mut guard = self.inner.lock().unwrap(); + let media = guard.as_mut().ok_or(MoqError::Closed)?; + media.import.set_enabled(enabled).map_err(|err| MoqError::Codec { + reason: format!("set enabled failed: {err}"), + })?; + Ok(()) + } + /// Finalize the track, publishing the last access unit still waiting for a following start code. /// /// The broadcast keeps the finished track until it ends, so its finish still holds it. A diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index c055599ad6b..55f6d9f41b6 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -2925,7 +2925,7 @@ async fn video_rung_sits_below_its_lower_and_disables() { &broadcast, named("high"), MoqVideoInit { - lower: Some(low.clone()), + lower: Some(MoqMediaProducer::Track { producer: low.clone() }), ..video_init(MoqVideoFormat::Avc3, h264_init()) }, ) @@ -2953,6 +2953,98 @@ async fn video_rung_sits_below_its_lower_and_disables() { assert!(disabled.video["low"].enabled); } +/// A byte-stream rung above another sits one priority below it, so a grant short of both funds +/// the lower one first. Disabled, it drops what it is written, and enabled again it resumes at +/// the next keyframe in the stream. +#[tokio::test] +async fn stream_rung_is_funded_after_its_lower_and_resumes_at_a_keyframe() { + let estimate = moq_net::bandwidth::Producer::new(); + let bandwidth = crate::bandwidth::MoqAllocator::new(moq_net::bandwidth::Allocator::new(estimate.consume())); + let broadcast = MoqBroadcastProducer::new().unwrap(); + let consumer = broadcast.consume().unwrap(); + let catalog_consumer = MoqMediaCatalogConsumer::subscribe(&consumer).await.unwrap(); + + let named = |name: &str| MoqMediaTarget::Named { + name: Some(name.into()), + }; + let low = MoqMediaTrackStreamProducer::video(&broadcast, named("low"), video_init(MoqVideoFormat::Avc3, h264_init())) + .unwrap(); + let high = MoqMediaTrackStreamProducer::video( + &broadcast, + named("high"), + MoqVideoInit { + lower: Some(MoqMediaProducer::Stream { producer: low.clone() }), + ..video_init(MoqVideoFormat::Avc3, h264_init()) + }, + ) + .unwrap(); + + let low_track = consumer.subscribe_track("low".into(), None).await.unwrap(); + let high_track = consumer.subscribe_track("high".into(), None).await.unwrap(); + assert_eq!(low_track.info().priority, hang::catalog::PRIORITY.video); + assert_eq!(high_track.info().priority, hang::catalog::PRIORITY.video - 1); + + // The higher rung asks for more, but strict priority fills the lower one first. + let low_reservation = bandwidth.reserve(&low.demand().unwrap(), 2_000_000); + let high_reservation = bandwidth.reserve(&high.demand().unwrap(), 4_000_000); + estimate + .set(Some(moq_net::bandwidth::Rate::from_bps(3_000_000))) + .unwrap(); + assert_eq!(low_reservation.rate(), Some(2_000_000)); + assert_eq!(high_reservation.rate(), Some(1_000_000)); + + // Each keyframe ends in its own slice bytes, to tell which one went out. + let keyframe = |last: u8| { + let mut bytes = h264_init(); + bytes.extend_from_slice(&[0, 0, 0, 1, 0x65, 0x88, 0x84, last]); + bytes + }; + let delta = [0, 0, 0, 1, 0x41, 0x9a, 0x21, 0x22]; + // The splitter emits an access unit once the next one starts, hence the trailing deltas. + let write = |bytes: &[&[u8]]| { + for bytes in bytes { + high.write(bytes.to_vec()).unwrap(); + } + }; + let next_frame = async || { + loop { + let frame = kio::clock::timeout(TIMEOUT, high_track.read_frame()) + .await + .expect("timed out waiting for a frame") + .unwrap() + .expect("the track ended"); + // The discontinuity marker is empty. + if !frame.payload.is_empty() { + return frame.payload; + } + } + }; + let high_enabled = async |enabled: bool| loop { + let catalog = kio::clock::timeout(TIMEOUT, catalog_consumer.next()) + .await + .expect("timed out waiting for catalog") + .unwrap() + .expect("expected a catalog"); + if catalog.video.get("high").is_some_and(|high| high.enabled == enabled) { + assert!(catalog.video["low"].enabled); + break; + } + }; + + write(&[&keyframe(0x21), &delta, &delta]); + assert!(next_frame().await.ends_with(&[0x21])); + + high.set_enabled(false).unwrap(); + high_enabled(false).await; + // The stream keeps flowing, but none of it goes out, the keyframe included. + write(&[&delta, &keyframe(0x23), &delta, &delta]); + + high.set_enabled(true).unwrap(); + write(&[&delta, &keyframe(0x24), &delta, &delta]); + high_enabled(true).await; + assert!(next_frame().await.ends_with(&[0x24]), "resumed on another keyframe"); +} + /// 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] diff --git a/rs/moq-mux/src/import/track.rs b/rs/moq-mux/src/import/track.rs index dd63d255a6d..a01c2866411 100644 --- a/rs/moq-mux/src/import/track.rs +++ b/rs/moq-mux/src/import/track.rs @@ -8,6 +8,7 @@ use crate::Result; use crate::catalog::VideoHint; use crate::catalog::hang::CatalogExt; +use crate::container::Frame; use super::{AudioFormat, VideoFormat}; pub use super::{AudioInit, VideoInit}; @@ -591,6 +592,55 @@ macro_rules! stream { /// H.264, an fMP4 reader, …); the importer infers them. pub struct TrackStream { kind: TrackStreamKind, + gate: Gate, +} + +/// Which of the access units a [`TrackStream`] splits it publishes. +/// +/// The stream's caller cannot see frame boundaries, so the importer, not the caller, drops a +/// disabled rendition's access units and finds the keyframe that resumes it. +#[derive(Default)] +struct Gate { + state: GateState, + /// Whether an access unit went out, so a disable has a timeline to break. + published: bool, +} + +#[derive(Clone, Copy, Default, PartialEq, Eq)] +enum GateState { + #[default] + Open, + /// Disabled: every access unit is dropped. + Closed, + /// Enabled again, but still disabled in the catalog until the next keyframe. + Resuming, +} + +impl Gate { + /// The access units of `frames` to publish, and whether the first of them resumes the + /// rendition, so the catalog enables it again. + fn admit(&mut self, mut frames: Vec) -> (Vec, bool) { + let resumed = match self.state { + GateState::Open => false, + GateState::Closed => { + frames.clear(); + false + } + GateState::Resuming => match frames.iter().position(|frame| frame.keyframe) { + Some(keyframe) => { + frames.drain(..keyframe); + self.state = GateState::Open; + true + } + None => { + frames.clear(); + false + } + }, + }; + self.published |= !frames.is_empty(); + (frames, resumed) + } } impl TrackStream { @@ -626,7 +676,10 @@ impl TrackStream { format => return Err(crate::Error::NotSelfDescribing(format.to_string())), }; - let mut stream = Self { kind }; + let mut stream = Self { + kind, + gate: Gate::default(), + }; if !init.data.is_empty() { stream.initialize(&init.data)?; } @@ -643,7 +696,10 @@ impl TrackStream { ref mut import, } => { import.initialize(data)?; - let frames = split.decode(data, None)?; + let (frames, resumed) = self.gate.admit(split.decode(data, None)?); + if resumed { + import.set_enabled(true)?; + } import.decode(frames)?; } TrackStreamKind::Hev1 { @@ -651,7 +707,10 @@ impl TrackStream { ref mut import, } => { import.initialize(data)?; - let frames = split.decode(data, None)?; + let (frames, resumed) = self.gate.admit(split.decode(data, None)?); + if resumed { + import.set_enabled(true)?; + } import.decode(frames)?; } TrackStreamKind::Av01 { @@ -666,6 +725,10 @@ impl TrackStream { } else { split.decode(data, None)? }; + let (frames, resumed) = self.gate.admit(frames); + if resumed { + import.set_enabled(true)?; + } import.decode(frames)?; } } @@ -676,7 +739,10 @@ impl TrackStream { /// Decode a chunk of the byte stream. pub fn decode(&mut self, data: &[u8]) -> Result<()> { stream!(&mut self.kind, split, import => { - let frames = split.decode(data, None)?; + let (frames, resumed) = self.gate.admit(split.decode(data, None)?); + if resumed { + import.set_enabled(true)?; + } import.decode(frames) }) } @@ -686,12 +752,46 @@ impl TrackStream { // The splitter is taken out to flush, so its input storage ends with the input. The import // stays, keeping the finished track and its rendition published. stream!(&mut self.kind, split, import => { - let tail = std::mem::take(split).flush(None)?; + let (tail, resumed) = self.gate.admit(std::mem::take(split).flush(None)?); + if resumed { + import.set_enabled(true)?; + } import.decode(tail)?; import.finish() }) } + /// Stop or resume publishing this rendition, advertised as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled). + /// + /// Disable it when its sender cannot be funded (a + /// [`rate::Target::Disabled`](crate::rate::Target::Disabled)). Once anything was published, + /// that marks a [`discontinuity`](crate::container::Producer::discontinuity). Every access + /// unit decoded until it is enabled again is dropped, so the caller may keep writing the + /// stream. Enabled again, it drops access units until the next keyframe and advertises + /// itself enabled with that keyframe. + pub fn set_enabled(&mut self, enabled: bool) -> Result<()> { + match (enabled, self.gate.state) { + (false, GateState::Open) => { + let published = self.gate.published; + stream!(&mut self.kind, split, import => { + import.set_enabled(false)?; + if published { + // The partial access unit belongs before the break, and would otherwise + // complete with the first bytes after it. + split.reset(); + import.discontinuity()?; + } + }); + self.gate.state = GateState::Closed; + } + (false, GateState::Resuming) => self.gate.state = GateState::Closed, + (true, GateState::Closed) => self.gate.state = GateState::Resuming, + (false, GateState::Closed) | (true, GateState::Open | GateState::Resuming) => {} + } + Ok(()) + } + /// Abort the track with `err` instead of finishing it cleanly, so subscribers /// see the real cause rather than [`moq_net::Error::Dropped`]. Consumes the importer. pub fn abort(self, err: moq_net::Error) { @@ -1510,4 +1610,66 @@ mod tests { ); } } + + /// A disabled [`TrackStream`] drops what it splits and marks the break, so a caller that + /// keeps writing sends nothing. Enabled again, it resumes, and advertises itself enabled, + /// only at the next keyframe. + #[tokio::test(start_paused = true)] + async fn track_stream_disables_and_resumes_at_a_keyframe() { + const H264_P: &[u8] = &[0x41, 0x9a, 0x21, 0x22]; + + let (broadcast, catalog) = new_broadcast(); + let consumer = broadcast.consume(); + let request = broadcast.reserve_track("video").unwrap(); + let mut import = TrackStream::video( + request, + catalog.reserve(), + VideoInit::new(VideoFormat::Avc3, Vec::new()), + ) + .unwrap(); + // Every group is written before anything reads, which the default + // REAL_TIME budget would collapse to the live edge. + let subscriber = consumer + .track("video") + .unwrap() + .subscribe( + moq_net::track::Subscription::default() + .with_delay(moq_net::track::Delay::default().with_max(Duration::from_secs(30))), + ) + .await + .unwrap(); + let enabled = || catalog.snapshot().video.renditions["video"].enabled; + + let keyframe = [h264_init(), annexb(&[H264_IDR])].concat(); + let delta = annexb(&[H264_P]); + // One access unit per frame interval. The splitter holds the last NAL until the next + // start code, so an access unit goes out one write after its last NAL arrives. + async fn write(import: &mut TrackStream, bytes: &[u8]) { + import.decode(bytes).unwrap(); + kio::clock::sleep(Duration::from_millis(33)).await; + } + for bytes in [&keyframe, &delta, &delta, &delta] { + write(&mut import, bytes).await; + } + + import.set_enabled(false).unwrap(); + assert!(!enabled()); + for bytes in [&delta, &keyframe, &delta, &delta] { + write(&mut import, bytes).await; + } + + import.set_enabled(true).unwrap(); + for bytes in [&delta, &keyframe, &delta] { + write(&mut import, bytes).await; + } + assert!(!enabled(), "enabled before the keyframe that resumes it"); + write(&mut import, &delta).await; + assert!(enabled()); + import.finish().unwrap(); + + // The first group loses the access units still in the splitter at the disable, the + // marker breaks the timeline, and the resumed group opens on the keyframe: three + // access units and the end marker `finish` bounds the last one with. + assert_eq!(collect_groups(subscriber).await, vec![2, 1, 4]); + } } From 9f709d3cfb0bba99df877ec7cf9c31bfbf978c2a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:18:46 -0800 Subject: [PATCH 3/8] feat(bindings)!: pass byte-stream producers as a lower rendition Co-Authored-By: Claude Opus 5.5 --- cpp/moq/MIGRATE.md | 4 + cpp/obs/src/moq-output.cpp | 2 +- dart/moq/MIGRATE.md | 4 + dart/moq/lib/media.dart | 3 + dart/moq_ffi/lib/src/moq.dart | 245 ++++++++++++++---- doc/lib/cpp/index.md | 2 +- doc/lib/dart/index.md | 2 +- doc/lib/go/index.md | 2 +- doc/lib/kt/index.md | 2 +- doc/lib/py/index.md | 2 +- doc/lib/swift/index.md | 2 +- go/wrapper/MIGRATE.md | 4 + go/wrapper/media/media.go | 43 ++- kt/moq/MIGRATE.md | 4 + .../kotlin/dev/moq/media/Media.kt | 3 + py/moq-rs/MIGRATE.md | 4 + py/moq-rs/moq/media.py | 27 +- quest/m1/README.md | 1 - quest/m1/stream-ladder.md | 35 --- rs/moq-ffi/MIGRATE.md | 4 + swift/MIGRATE.md | 4 + swift/Sources/Moq/Media.swift | 32 ++- 22 files changed, 324 insertions(+), 107 deletions(-) delete mode 100644 quest/m1/stream-ladder.md diff --git a/cpp/moq/MIGRATE.md b/cpp/moq/MIGRATE.md index d561d680531..18b1fd1e0bf 100644 --- a/cpp/moq/MIGRATE.md +++ b/cpp/moq/MIGRATE.md @@ -2,6 +2,10 @@ ## Unreleased +### `moq::VideoInit::lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`lower` is `std::optional` instead of a `std::shared_ptr`. Assign `moq::MediaProducer(moq::MediaProducer::kTrack{producer})`, or `kStream` for a `moq::MediaTrackStreamProducer`. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `moq::Announce.captures` is `std::vector` instead of an optional vector. Remove optional checks and dereferences; an empty vector means the filter has no wildcards. Regenerate bindings and rebuild with the matching native library. diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 29ad0eaab46..95bfb576647 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -850,7 +850,7 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) if (below == video_tracks.end()) return; if (below->second.producer) { - config.lower = below->second.producer; + config.lower = moq::MediaProducer(moq::MediaProducer::kTrack{below->second.producer}); lower_bps = (uint64_t)lower_kbps * 1000ULL; } } diff --git a/dart/moq/MIGRATE.md b/dart/moq/MIGRATE.md index 38f4d5f1f5b..9dd228b6e21 100644 --- a/dart/moq/MIGRATE.md +++ b/dart/moq/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `VideoInit.lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`VideoInit.lower` is a `Producer` (`MoqMediaProducer`) instead of a `TrackProducer`. Pass `TrackMoqMediaProducer(producer: ...)`, or `StreamMoqMediaProducer(producer: ...)` for a byte-stream rendition. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `Announce.captures` is `List` instead of `List?`. Remove null checks and pass `captures: []` when constructing an announcement for a filter without wildcards. Regenerate bindings and rebuild with the matching native library. diff --git a/dart/moq/lib/media.dart b/dart/moq/lib/media.dart index 674d24c0a79..53d8ff853bc 100644 --- a/dart/moq/lib/media.dart +++ b/dart/moq/lib/media.dart @@ -9,6 +9,9 @@ typedef TrackProducer = MoqMediaTrackProducer; /// The write side of a media track fed a raw byte stream, with frame boundaries inferred. typedef TrackStreamProducer = MoqMediaTrackStreamProducer; +/// A single-track producer of either kind, such as a video rendition's `lower`. +typedef Producer = MoqMediaProducer; + /// The write side of a container, which publishes each track it describes. typedef ContainerProducer = MoqMediaContainerProducer; diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 469ee931747..c056bdef850 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -2546,7 +2546,7 @@ class MoqVideoInit { final Uint8List data; final String? label; final MoqVideoHint? hint; - final MoqMediaTrackProducer? lower; + final MoqMediaProducer? lower; MoqVideoInit({ required this.format, required this.data, @@ -2583,7 +2583,7 @@ class FfiConverterMoqVideoInit { ); final hint = hint_lifted.value; new_offset += hint_lifted.bytesRead; - final lower_lifted = FfiConverterOptionalMoqMediaTrackProducer.read( + final lower_lifted = FfiConverterOptionalMoqMediaProducer.read( Uint8List.view(buf.buffer, new_offset), ); final lower = lower_lifted.value; @@ -2606,7 +2606,7 @@ class FfiConverterMoqVideoInit { FfiConverterUint8List.allocationSize(value.data) + FfiConverterOptionalString.allocationSize(value.label) + FfiConverterOptionalMoqVideoHint.allocationSize(value.hint) + - FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value.lower) + + FfiConverterOptionalMoqMediaProducer.allocationSize(value.lower) + 0; final buf = Uint8List(total_length); write(value, buf); @@ -2631,7 +2631,7 @@ class FfiConverterMoqVideoInit { value.hint, Uint8List.view(buf.buffer, new_offset), ); - new_offset += FfiConverterOptionalMoqMediaTrackProducer.write( + new_offset += FfiConverterOptionalMoqMediaProducer.write( value.lower, Uint8List.view(buf.buffer, new_offset), ); @@ -2643,7 +2643,7 @@ class FfiConverterMoqVideoInit { FfiConverterUint8List.allocationSize(value.data) + FfiConverterOptionalString.allocationSize(value.label) + FfiConverterOptionalMoqVideoHint.allocationSize(value.hint) + - FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value.lower) + + FfiConverterOptionalMoqMediaProducer.allocationSize(value.lower) + 0; } } @@ -7568,6 +7568,130 @@ class FfiConverterMoqDecodeStart { } } +abstract class MoqMediaProducer { + RustBuffer lower(); + int allocationSize(); + int write(Uint8List buf); +} + +class FfiConverterMoqMediaProducer { + static MoqMediaProducer lift(RustBuffer buffer) { + return FfiConverterMoqMediaProducer.read(buffer.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + final index = buf.buffer.asByteData(buf.offsetInBytes).getInt32(0); + final subview = Uint8List.view(buf.buffer, buf.offsetInBytes + 4); + switch (index) { + case 1: + final lifted = TrackMoqMediaProducer.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + case 2: + final lifted = StreamMoqMediaProducer.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + default: + throw UniffiInternalError( + UniffiInternalError.unexpectedEnumCase, + "Unable to determine enum variant", + ); + } + } + + static RustBuffer lower(MoqMediaProducer value) { + return value.lower(); + } + + static int allocationSize(MoqMediaProducer value) { + return value.allocationSize(); + } + + static int write(MoqMediaProducer value, Uint8List buf) { + return value.write(buf) - buf.offsetInBytes; + } +} + +class TrackMoqMediaProducer extends MoqMediaProducer { + final MoqMediaTrackProducer producer; + TrackMoqMediaProducer(MoqMediaTrackProducer this.producer); + TrackMoqMediaProducer._(MoqMediaTrackProducer this.producer); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final producer_lifted = FfiConverterMoqMediaTrackProducer.read( + Uint8List.view(buf.buffer, new_offset), + ); + final producer = producer_lifted.value; + new_offset += producer_lifted.bytesRead; + return LiftRetVal(TrackMoqMediaProducer._(producer), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterMoqMediaTrackProducer.allocationSize(producer) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 1); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterMoqMediaTrackProducer.write( + producer, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + +class StreamMoqMediaProducer extends MoqMediaProducer { + final MoqMediaTrackStreamProducer producer; + StreamMoqMediaProducer(MoqMediaTrackStreamProducer this.producer); + StreamMoqMediaProducer._(MoqMediaTrackStreamProducer this.producer); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final producer_lifted = FfiConverterMoqMediaTrackStreamProducer.read( + Uint8List.view(buf.buffer, new_offset), + ); + final producer = producer_lifted.value; + new_offset += producer_lifted.bytesRead; + return LiftRetVal(StreamMoqMediaProducer._(producer), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterMoqMediaTrackStreamProducer.allocationSize(producer) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 2); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterMoqMediaTrackStreamProducer.write( + producer, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + abstract class MoqMediaRead { RustBuffer lower(); int allocationSize(); @@ -12138,6 +12262,7 @@ class FfiConverterMoqMediaTrackProducer { abstract class MoqMediaTrackStreamProducerInterface { MoqTrackDemand demand(); void finish(); + void setEnabled({required bool enabled}); void write({required Uint8List payload}); } @@ -12218,6 +12343,16 @@ class MoqMediaTrackStreamProducer }, moqExceptionErrorHandler); } + void setEnabled({required bool enabled}) { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqmediatrackstreamproducer_set_enabled( + uniffiClonePointer(), + FfiConverterBool.lower(enabled), + status, + ); + }, moqExceptionErrorHandler); + } + void write({required Uint8List payload}) { return rustCall((status) { uniffi_moq_ffi_fn_method_moqmediatrackstreamproducer_write( @@ -16613,46 +16748,46 @@ class FfiConverterOptionalMoqMediaFrame { } } -class FfiConverterOptionalMoqMediaRead { - static MoqMediaRead? lift(RustBuffer buf) { - return FfiConverterOptionalMoqMediaRead.read(buf.asUint8List()).value; +class FfiConverterOptionalMoqMediaProducer { + static MoqMediaProducer? lift(RustBuffer buf) { + return FfiConverterOptionalMoqMediaProducer.read(buf.asUint8List()).value; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { return LiftRetVal(null, 1); } - final result = FfiConverterMoqMediaRead.read( + final result = FfiConverterMoqMediaProducer.read( Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ); - return LiftRetVal(result.value, result.bytesRead + 1); + return LiftRetVal(result.value, result.bytesRead + 1); } - static int allocationSize([MoqMediaRead? value]) { + static int allocationSize([MoqMediaProducer? value]) { if (value == null) { return 1; } - return FfiConverterMoqMediaRead.allocationSize(value) + 1; + return FfiConverterMoqMediaProducer.allocationSize(value) + 1; } - static RustBuffer lower(MoqMediaRead? value) { + static RustBuffer lower(MoqMediaProducer? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqMediaRead.allocationSize(value), + FfiConverterOptionalMoqMediaProducer.allocationSize(value), ); - FfiConverterOptionalMoqMediaRead.write(value, buf); + FfiConverterOptionalMoqMediaProducer.write(value, buf); return toRustBuffer(buf); } - static int write(MoqMediaRead? value, Uint8List buf) { + static int write(MoqMediaProducer? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqMediaRead.write( + return FfiConverterMoqMediaProducer.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -16660,46 +16795,46 @@ class FfiConverterOptionalMoqMediaRead { } } -class FfiConverterOptionalMoqMediaTimebase { - static MoqMediaTimebase? lift(RustBuffer buf) { - return FfiConverterOptionalMoqMediaTimebase.read(buf.asUint8List()).value; +class FfiConverterOptionalMoqMediaRead { + static MoqMediaRead? lift(RustBuffer buf) { + return FfiConverterOptionalMoqMediaRead.read(buf.asUint8List()).value; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { return LiftRetVal(null, 1); } - final result = FfiConverterMoqMediaTimebase.read( + final result = FfiConverterMoqMediaRead.read( Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ); - return LiftRetVal(result.value, result.bytesRead + 1); + return LiftRetVal(result.value, result.bytesRead + 1); } - static int allocationSize([MoqMediaTimebase? value]) { + static int allocationSize([MoqMediaRead? value]) { if (value == null) { return 1; } - return FfiConverterMoqMediaTimebase.allocationSize(value) + 1; + return FfiConverterMoqMediaRead.allocationSize(value) + 1; } - static RustBuffer lower(MoqMediaTimebase? value) { + static RustBuffer lower(MoqMediaRead? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqMediaTimebase.allocationSize(value), + FfiConverterOptionalMoqMediaRead.allocationSize(value), ); - FfiConverterOptionalMoqMediaTimebase.write(value, buf); + FfiConverterOptionalMoqMediaRead.write(value, buf); return toRustBuffer(buf); } - static int write(MoqMediaTimebase? value, Uint8List buf) { + static int write(MoqMediaRead? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqMediaTimebase.write( + return FfiConverterMoqMediaRead.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -16707,51 +16842,46 @@ class FfiConverterOptionalMoqMediaTimebase { } } -class FfiConverterOptionalMoqMediaTrackProducer { - static MoqMediaTrackProducer? lift(RustBuffer buf) { - return FfiConverterOptionalMoqMediaTrackProducer.read( - buf.asUint8List(), - ).value; +class FfiConverterOptionalMoqMediaTimebase { + static MoqMediaTimebase? lift(RustBuffer buf) { + return FfiConverterOptionalMoqMediaTimebase.read(buf.asUint8List()).value; } - static LiftRetVal read(Uint8List buf) { + static LiftRetVal read(Uint8List buf) { if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { return LiftRetVal(null, 1); } - final result = FfiConverterMoqMediaTrackProducer.read( + final result = FfiConverterMoqMediaTimebase.read( Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ); - return LiftRetVal( - result.value, - result.bytesRead + 1, - ); + return LiftRetVal(result.value, result.bytesRead + 1); } - static int allocationSize([MoqMediaTrackProducer? value]) { + static int allocationSize([MoqMediaTimebase? value]) { if (value == null) { return 1; } - return FfiConverterMoqMediaTrackProducer.allocationSize(value) + 1; + return FfiConverterMoqMediaTimebase.allocationSize(value) + 1; } - static RustBuffer lower(MoqMediaTrackProducer? value) { + static RustBuffer lower(MoqMediaTimebase? value) { if (value == null) { return toRustBuffer(Uint8List.fromList([0])); } final buf = Uint8List( - FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value), + FfiConverterOptionalMoqMediaTimebase.allocationSize(value), ); - FfiConverterOptionalMoqMediaTrackProducer.write(value, buf); + FfiConverterOptionalMoqMediaTimebase.write(value, buf); return toRustBuffer(buf); } - static int write(MoqMediaTrackProducer? value, Uint8List buf) { + static int write(MoqMediaTimebase? value, Uint8List buf) { if (value == null) { buf[0] = 0; return 1; } buf[0] = 1; - return FfiConverterMoqMediaTrackProducer.write( + return FfiConverterMoqMediaTimebase.write( value, Uint8List.view(buf.buffer, buf.offsetInBytes + 1), ) + @@ -19395,6 +19525,15 @@ external void uniffi_moq_ffi_fn_method_moqmediatrackstreamproducer_finish( Pointer uniffiStatus, ); +@Native, Int8, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqmediatrackstreamproducer_set_enabled( + Pointer ptr, + int enabled, + Pointer uniffiStatus, +); + @Native, RustBuffer, Pointer)>( assetId: _uniffiAssetId, ) @@ -21354,6 +21493,10 @@ uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_demand(); external int uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_finish(); +@Native(assetId: _uniffiAssetId) +external int +uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_set_enabled(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_write(); @@ -22079,6 +22222,10 @@ void _checkApiChecksums() { 43608) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_set_enabled() != + 9270) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqmediatrackstreamproducer_write() != 44842) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index 4b139d69663..c6e0b9085ee 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -149,7 +149,7 @@ ok(publication->announce({})); // each broadcast mints a fresh instance ok(broadcast->close()); // keep the producer alive while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(false)` after a `discontinuity()`, and `set_enabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `moq::MediaProducer::kTrack` or `kStream`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(false)` after a `discontinuity()`, and `set_enabled(true)` on the keyframe that resumes it. A `MediaTrackStreamProducer` finds its own frames: `set_enabled(false)` drops what it is written, and after `set_enabled(true)` it resumes at the stream's next keyframe. ## Futures diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 106c767cb72..352d2f5859a 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -139,7 +139,7 @@ mine.close(); await moq.close(); ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `TrackMoqMediaProducer` or `StreamMoqMediaProducer`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what it is written, and after `setEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index 5ad9e4eb92e..a80fe99afee 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -119,7 +119,7 @@ _ = video.Finish() broadcast.Close() // keep the producer reachable while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `Lower` in its `media.VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `SetEnabled(false)` after a `Discontinuity()`, and `SetEnabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `Lower` in its `media.VideoInit` too, a `*media.TrackProducer` or `*media.TrackStreamProducer`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `SetEnabled(false)` after a `Discontinuity()`, and `SetEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `SetEnabled(false)` drops what it is written, and after `SetEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index d7c33fb9fe5..a56e6a6e952 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -94,7 +94,7 @@ Moq.connect("https://relay.example.com").use { moq -> } ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `uniffi.moq.MoqMediaProducer.Track` or `.Stream`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what it is written, and after `setEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 3f1a8d5ae3f..064f9f0171d 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -109,7 +109,7 @@ async def main(): asyncio.run(main()) ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `media.TrackProducer.video` takes `lower` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(False)` after a `discontinuity()`, and `set_enabled(True)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `media.TrackProducer.video` or `media.TrackStreamProducer.video` takes `lower` too, of either kind, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(False)` after a `discontinuity()`, and `set_enabled(True)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `set_enabled(False)` drops what it is written, and after `set_enabled(True)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index ec0ae5b804a..01b98031473 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -95,7 +95,7 @@ session.shutdown() try await session.closed() ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `Media.TrackProducer.video` takes `lower` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `Media.TrackProducer.video` or `Media.TrackStreamProducer.video` takes `lower` too, as a `Media.Producer` of either kind, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what it is written, and after `setEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index 8f0b8385259..e44ae42a4ca 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `media.VideoInit.Lower` is a `media.Producer` ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`Lower` is the `media.Producer` interface instead of `*media.TrackProducer`, so a `*media.TrackStreamProducer` fits too. Assigning a `*media.TrackProducer`, nil included, compiles unchanged; code that reads `Lower` back type-asserts it. + ### `media.VideoInit` is a struct `media.VideoInit` is its own struct rather than an alias of the generated record, so its new `Lower` field takes a `*media.TrackProducer`. A composite literal with `Format`, `Data`, `Label`, and `Hint` compiles unchanged; build one that way where you converted a generated `MoqVideoInit`. diff --git a/go/wrapper/media/media.go b/go/wrapper/media/media.go index bbee93deab5..0c7e32d6cb3 100644 --- a/go/wrapper/media/media.go +++ b/go/wrapper/media/media.go @@ -73,15 +73,25 @@ type VideoInit struct { Label *string // Hint carries catalog fields the stream cannot reveal itself. Hint *VideoHint - // Lower is the next lower rendition of the same source: this track then sits one - // priority below it, so a bandwidth allocator funds that one first. - Lower *TrackProducer + // Lower is the next lower rendition of the same source, a *TrackProducer or a + // *TrackStreamProducer: this track then sits one priority below it, so a bandwidth + // allocator funds that one first. + Lower Producer +} + +// Producer is a single-track producer of either kind, *TrackProducer or +// *TrackStreamProducer, such as a video rendition's Lower. +type Producer interface { + // producer is nil for a nil pointer, which means no producer, as a nil Lower does. + producer() ffi.MoqMediaProducer } func (init VideoInit) ffi() ffi.MoqVideoInit { - var lower **ffi.MoqMediaTrackProducer + var lower *ffi.MoqMediaProducer if init.Lower != nil { - lower = &init.Lower.inner + if producer := init.Lower.producer(); producer != nil { + lower = &producer + } } return ffi.MoqVideoInit{Format: init.Format, Data: init.Data, Label: init.Label, Hint: init.Hint, Lower: lower} } @@ -443,6 +453,13 @@ func (m *TrackProducer) Finish() error { return m.inner.Finish() } +func (m *TrackProducer) producer() ffi.MoqMediaProducer { + if m == nil { + return nil + } + return ffi.MoqMediaProducerTrack{Producer: m.inner} +} + // ContainerProducer publishes a container, which demuxes and publishes its own // tracks. Unlike TrackProducer there is no per-frame timestamp: a container // carries its tracks' timing itself. @@ -492,11 +509,27 @@ func (m *TrackStreamProducer) Write(payload []byte) error { return m.inner.Write(payload) } +// SetEnabled stops or resumes publishing this rendition, advertised as its catalog +// enabled flag. Disable it when its sender stops for lack of bandwidth: this marks a +// discontinuity and drops everything written until it is enabled again, so the stream +// may keep flowing. Enabled again, it resumes, and advertises itself enabled, at the +// next keyframe in the stream. +func (m *TrackStreamProducer) SetEnabled(enabled bool) error { + return m.inner.SetEnabled(enabled) +} + // Finish closes the stream. func (m *TrackStreamProducer) Finish() error { return m.inner.Finish() } +func (m *TrackStreamProducer) producer() ffi.MoqMediaProducer { + if m == nil { + return nil + } + return ffi.MoqMediaProducerStream{Producer: m.inner} +} + // Read is one encoded frame, a ReadGap, or a ReadEnd, in delivery order. type Read interface{ read() } diff --git a/kt/moq/MIGRATE.md b/kt/moq/MIGRATE.md index ecb3c831ba8..147c40a72a1 100644 --- a/kt/moq/MIGRATE.md +++ b/kt/moq/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `VideoInit.lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`VideoInit.lower` is a `dev.moq.media.Producer` (`uniffi.moq.MoqMediaProducer`) instead of a `TrackProducer`. Pass `MoqMediaProducer.Track(producer)`, or `MoqMediaProducer.Stream(producer)` for a byte-stream rendition. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `Announce.captures` is `List` instead of `List?`. Remove null checks and pass `captures = emptyList()` when constructing an announcement for a filter without wildcards. Rebuild with the matching generated bindings and native library. diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt index 37037d8f006..ef44715c3e2 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/media/Media.kt @@ -13,6 +13,9 @@ typealias TrackProducer = uniffi.moq.MoqMediaTrackProducer /** The write side of a media track fed a raw byte stream, with frame boundaries inferred. */ typealias TrackStreamProducer = uniffi.moq.MoqMediaTrackStreamProducer +/** A single-track producer of either kind, such as a video rendition's `lower`. */ +typealias Producer = uniffi.moq.MoqMediaProducer + /** The write side of a container, which publishes each track it describes. */ typealias ContainerProducer = uniffi.moq.MoqMediaContainerProducer diff --git a/py/moq-rs/MIGRATE.md b/py/moq-rs/MIGRATE.md index af31a32d54a..ed54a682a2f 100644 --- a/py/moq-rs/MIGRATE.md +++ b/py/moq-rs/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `VideoInit.lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +The `lower=` argument of `TrackProducer.video` and `TrackStreamProducer.video` accepts either producer and needs no change. Code that sets the raw `VideoInit.lower` field wraps the handle as `MoqMediaProducer.TRACK(producer=...)` or `MoqMediaProducer.STREAM(producer=...)`. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `Announce.captures` is `list[str]` instead of `list[str] | None`. Remove `None` checks and pass `captures=[]` when constructing an announcement for a filter without wildcards. Rebuild with the matching generated bindings and native library. diff --git a/py/moq-rs/moq/media.py b/py/moq-rs/moq/media.py index 106277a431b..b0105aaec62 100644 --- a/py/moq-rs/moq/media.py +++ b/py/moq-rs/moq/media.py @@ -30,6 +30,7 @@ MoqMediaFrame, MoqMediaGroupConfig, MoqMediaGroupConsumer, + MoqMediaProducer, MoqMediaRead, MoqMediaTarget, MoqMediaTimebase, @@ -174,12 +175,13 @@ def video( init: VideoInit, *, target: Target = Named(), - lower: TrackProducer | None = None, + lower: TrackProducer | TrackStreamProducer | None = None, ) -> TrackProducer: """Import complete encoded video frames on a named or requested track. - Pass ``lower``, the next lower rendition of the same source, to publish this - track one priority below it, so a bandwidth allocator funds that one first. + Pass ``lower``, the next lower rendition of the same source and of either + producer kind, to publish this track one priority below it, so a bandwidth + allocator funds that one first. """ return cls(MoqMediaTrackProducer.video(broadcast._inner, target._ffi(), _lower(init, lower))) @@ -280,7 +282,7 @@ def video( init: VideoInit, *, target: Target = Named(), - lower: TrackProducer | None = None, + lower: TrackProducer | TrackStreamProducer | None = None, ) -> TrackStreamProducer: """Import a video byte stream, inferring frame boundaries. @@ -299,16 +301,29 @@ def write(self, payload: bytes) -> None: """Push raw stream bytes; whole frames are emitted as they complete.""" self._inner.write(payload) + def set_enabled(self, enabled: bool) -> None: + """Stop or resume publishing this rendition, advertised as its catalog ``enabled``. + + Disable it when its sender stops for lack of bandwidth: this marks a + discontinuity and drops everything written until it is enabled again, so + the stream may keep flowing. Enabled again, it resumes, and advertises + itself enabled, at the next keyframe in the stream. + """ + self._inner.set_enabled(enabled) + def finish(self) -> None: """Finish publishing and flush a clean end to subscribers.""" self._inner.finish() -def _lower(init: VideoInit, lower: TrackProducer | None) -> VideoInit: +def _lower(init: VideoInit, lower: TrackProducer | TrackStreamProducer | None) -> VideoInit: if lower is None: return init init = copy.copy(init) - init.lower = lower._inner + if isinstance(lower, TrackProducer): + init.lower = cast(MoqMediaProducer, MoqMediaProducer.TRACK(producer=lower._inner)) + else: + init.lower = cast(MoqMediaProducer, MoqMediaProducer.STREAM(producer=lower._inner)) return init diff --git a/quest/m1/README.md b/quest/m1/README.md index 3891b7870c4..bd257f9b3b8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -147,7 +147,6 @@ per-frame CPU growth since #5926. - [moq-c at parity](/quest/m1/c/README.md) - the hand-written `moq-c` ships at the cut with every m1 binding feature moq-ffi gains - [JSON writes take a broadcast-clock timestamp](/quest/m1/json-required-timestamp.md) - every binding's JSON write requires one, moq-c's NULL fails, and C gains `moq_publish_catalog_now()`, so C media stops advertising an 8 s delay - [Catalog delay is the least a track trails](/quest/m1/catalog-delay-min.md) - the estimator keeps the minimum trail, so B-frame reorder never becomes `delay` -- [Byte-stream renditions form a ladder](/quest/m1/stream-ladder.md) - moq-ffi stream producers link as `lower` and disable unfundable rungs like frame producers - [A job's exact route continues its claim's readers](/quest/m1/announce-takeover.md) - a job's route inherits its claim's instance, replacing the ANNOUNCE_TAKEOVER predecessor design - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges with absolute or relative ends, in either order, and replaces FETCH; merged ranges never starve a subscriber; relays fill misses by range, including over moq-transport - [Drop set](/quest/m1/drop-set.md) - each cursor judges missing groups against its own budget, so a narrow reader drops promptly and a widening re-offers groups already reported dropped on lite-07 diff --git a/quest/m1/stream-ladder.md b/quest/m1/stream-ladder.md deleted file mode 100644 index 0accc13b139..00000000000 --- a/quest/m1/stream-ladder.md +++ /dev/null @@ -1,35 +0,0 @@ -# [M] Byte-stream renditions form a ladder - -## Goal - -A simulcast ladder of moq-ffi byte-stream producers -(`MoqMediaTrackStreamProducer`, which imports fMP4, MKV, TS, or FLV bytes) -works like a ladder of frame producers (`MoqMediaTrackProducer`): - -- A stream producer can be passed as another rendition's `lower`, and sits - one priority below it. -- It disables itself when the bandwidth grant can't fund it, and resumes at - the next keyframe, as - [moq-dev/moq#7029](https://github.com/moq-dev/moq/pull/7029) does for - frame rungs. - -Every binding exposes this. - -## Plan - -Planned 2026-11-01 (maintainer, follow-up planning; m1 so the binding API -breaks before the cut). Raised by CodeRabbit on #7029 -([r4436515610](https://github.com/moq-dev/moq/pull/7029#discussion_r4436515610)). - -- The stream producer records its priority and exports `set_enabled`. -- `lower` (`MoqVideoInit.lower`, `MoqVideoProducerOptions.lower`) accepts - either producer kind. That changes a public UniFFI field type in every - binding (Python, Swift, Kotlin, Go, Dart). - - Pick the type that keeps one field, for example a small enum or a shared - rendition handle. - - Record the choice, and the migration in each `MIGRATE.md`. -- The C ABI follows when the C ABI parity work for #7029 does. -- Test a two-rung stream ladder in `rs/moq-ffi/src/test.rs`: the lower rung - is funded first, and the upper rung disables and resumes at a keyframe. - -Public API: breaking in every binding (`lower`'s type). Wire: none. diff --git a/rs/moq-ffi/MIGRATE.md b/rs/moq-ffi/MIGRATE.md index 7f6f48e6d25..f5a43dfcdfe 100644 --- a/rs/moq-ffi/MIGRATE.md +++ b/rs/moq-ffi/MIGRATE.md @@ -5,6 +5,10 @@ These are the UniFFI names; each language wrapper has its own `MIGRATE.md` in it ## Unreleased +### `MoqVideoInit.lower` takes either producer kind ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`MoqVideoInit.lower` is `Option` instead of `Option`. Wrap a frame producer as `MoqMediaProducer::Track { producer }`; a `MoqMediaTrackStreamProducer` goes in as `MoqMediaProducer::Stream { producer }`. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `MoqAnnounce.captures` is `Vec` instead of `Option>`. Remove optional checks and pass an empty list when constructing an announcement for a filter without wildcards. Regenerate bindings and rebuild the matching native library together. diff --git a/swift/MIGRATE.md b/swift/MIGRATE.md index 3ff5db0b9d9..242af8b38a0 100644 --- a/swift/MIGRATE.md +++ b/swift/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `lower` is a `Media.Producer` ([#7699](https://github.com/moq-dev/moq/pull/7699)) + +`Media.TrackProducer.video` and `Media.TrackStreamProducer.video` take `lower: Media.Producer?` instead of `TrackProducer?`. Pass `lower: .track(producer)`, or `.stream(producer)` for a byte-stream rendition. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `Announce.captures` is `[String]` instead of `[String]?`. Remove optional checks and pass `captures: []` when constructing an announcement for a filter without wildcards. Rebuild with the matching MoqFFI package and native library. diff --git a/swift/Sources/Moq/Media.swift b/swift/Sources/Moq/Media.swift index 80fdf822acf..3ff240ac7f9 100644 --- a/swift/Sources/Moq/Media.swift +++ b/swift/Sources/Moq/Media.swift @@ -53,6 +53,19 @@ public enum Media { } } } + /// A single-track producer of either kind, such as a video rendition's `lower`. + public enum Producer { + /// Imports complete encoded frames. + case track(TrackProducer) + /// Imports a byte stream. + case stream(TrackStreamProducer) + var ffi: MoqMediaProducer { + switch self { + case .track(let producer): return .track(producer: producer.ffi) + case .stream(let producer): return .stream(producer: producer.ffi) + } + } + } /// An audio codec, its initialization bytes, and label. public typealias AudioInit = MoqAudioInit /// A video codec, its initialization bytes, label, and hints. @@ -238,11 +251,12 @@ public enum Media { } /// Import complete video frames on a named or requested track. /// - /// Pass `lower`, the next lower rendition of the same source, to publish this track one - /// priority below it, so a bandwidth allocator funds that one first. + /// Pass `lower`, the next lower rendition of the same source and of either producer + /// kind, to publish this track one priority below it, so a bandwidth allocator funds + /// that one first. public static func video( broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil), - lower: TrackProducer? = nil + lower: Producer? = nil ) throws -> TrackProducer { var initData = initData if let lower { initData.lower = lower.ffi } @@ -351,7 +365,7 @@ public enum Media { /// `lower` is as for ``TrackProducer/video(broadcast:initData:target:lower:)``. public static func video( broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil), - lower: TrackProducer? = nil + lower: Producer? = nil ) throws -> TrackStreamProducer { var initData = initData if let lower { initData.lower = lower.ffi } @@ -373,6 +387,16 @@ public enum Media { try ffi.write(payload: payload) } + /// Stop or resume publishing this rendition, advertised as its catalog `enabled`. + /// + /// Disable it when its sender stops for lack of bandwidth: this marks a discontinuity + /// and drops everything written until it is enabled again, so the stream may keep + /// flowing. Enabled again, it resumes, and advertises itself enabled, at the next + /// keyframe in the stream. + public func setEnabled(_ enabled: Bool) throws { + try ffi.setEnabled(enabled: enabled) + } + /// Finalize the track, publishing the last access unit still waiting for a /// following start code. public func finish() throws { From 4fe91ca639c5ce9f52b3cca4dc7b552e77f5d929 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:21:19 -0800 Subject: [PATCH 4/8] style: rustfmt Co-Authored-By: Claude Opus 5.5 --- rs/moq-ffi/src/media.rs | 18 ++++++++++++++---- rs/moq-ffi/src/test.rs | 5 +++-- 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 057dbfe9c47..c2cb5bd2a64 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -424,11 +424,21 @@ impl MoqMediaProducer { match self { Self::Track { producer } => { let guard = producer.inner.lock().unwrap(); - guard.as_ref().ok_or(MoqError::Closed)?.video.ok_or_else(|| MoqError::Codec { - reason: "the lower rendition is not a video track".into(), - }) + guard + .as_ref() + .ok_or(MoqError::Closed)? + .video + .ok_or_else(|| MoqError::Codec { + reason: "the lower rendition is not a video track".into(), + }) } - Self::Stream { producer } => Ok(producer.inner.lock().unwrap().as_ref().ok_or(MoqError::Closed)?.priority), + Self::Stream { producer } => Ok(producer + .inner + .lock() + .unwrap() + .as_ref() + .ok_or(MoqError::Closed)? + .priority), } } } diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 55f6d9f41b6..0d0f913915f 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -2967,8 +2967,9 @@ async fn stream_rung_is_funded_after_its_lower_and_resumes_at_a_keyframe() { let named = |name: &str| MoqMediaTarget::Named { name: Some(name.into()), }; - let low = MoqMediaTrackStreamProducer::video(&broadcast, named("low"), video_init(MoqVideoFormat::Avc3, h264_init())) - .unwrap(); + let low = + MoqMediaTrackStreamProducer::video(&broadcast, named("low"), video_init(MoqVideoFormat::Avc3, h264_init())) + .unwrap(); let high = MoqMediaTrackStreamProducer::video( &broadcast, named("high"), From eec09447b770af666e5ece055c422acb76ed63ba Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:31:19 -0800 Subject: [PATCH 5/8] fix(mux): keep the byte-stream framing across a disable Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/import/track.rs | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/rs/moq-mux/src/import/track.rs b/rs/moq-mux/src/import/track.rs index a01c2866411..b80839a06c2 100644 --- a/rs/moq-mux/src/import/track.rs +++ b/rs/moq-mux/src/import/track.rs @@ -774,12 +774,12 @@ impl TrackStream { match (enabled, self.gate.state) { (false, GateState::Open) => { let published = self.gate.published; - stream!(&mut self.kind, split, import => { + // The splitter keeps its partial NAL, since the next chunk may continue it. The + // access unit in flight completes behind the gate, which drops it unless it is + // a keyframe that resumes the rendition. + stream!(&mut self.kind, _, import => { import.set_enabled(false)?; if published { - // The partial access unit belongs before the break, and would otherwise - // complete with the first bytes after it. - split.reset(); import.discontinuity()?; } }); @@ -1648,13 +1648,16 @@ mod tests { import.decode(bytes).unwrap(); kio::clock::sleep(Duration::from_millis(33)).await; } - for bytes in [&keyframe, &delta, &delta, &delta] { + for bytes in [&keyframe, &delta, &delta] { write(&mut import, bytes).await; } + // A chunk may end mid-NAL, and the next one continue it across the disable. + let (head, tail) = delta.split_at(6); + write(&mut import, head).await; import.set_enabled(false).unwrap(); assert!(!enabled()); - for bytes in [&delta, &keyframe, &delta, &delta] { + for bytes in [tail, &keyframe, &delta, &delta] { write(&mut import, bytes).await; } From fe803f3592b66d7034252a83c830154e3e5d6ffc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 09:34:23 -0800 Subject: [PATCH 6/8] fix(mux): a keyframe dropped while disabled still configures the importer Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/codec/av1/import.rs | 33 ++++---- rs/moq-mux/src/codec/h264/import.rs | 32 ++++---- rs/moq-mux/src/codec/h265/import.rs | 39 ++++++---- rs/moq-mux/src/import/track.rs | 114 +++++++++++++++++----------- 4 files changed, 132 insertions(+), 86 deletions(-) diff --git a/rs/moq-mux/src/codec/av1/import.rs b/rs/moq-mux/src/codec/av1/import.rs index 6d2d960b7b2..de57992d339 100644 --- a/rs/moq-mux/src/codec/av1/import.rs +++ b/rs/moq-mux/src/codec/av1/import.rs @@ -233,19 +233,7 @@ impl Import { /// keyframe's inline sequence header and refining the catalog jitter. fn write_frames(&mut self, frames: impl IntoIterator) -> Result<()> { for frame in frames { - if frame.keyframe { - let seq = find_sequence_header(&frame.payload); - // A new sequence header starts a coded sequence whose HDR metadata replaces the last - // one's. It goes first, so a keyframe that also configures publishes once. - // The first header after an av1C is the record's own sequence, which sync samples repeat. - let sequence = seq - .as_ref() - .is_some_and(|seq| self.last_seq.as_ref().is_some_and(|last| last != seq)); - self.catalog.hdr(&mut self.track, find_hdr(&frame.payload), sequence)?; - if let Some(seq) = seq { - self.configure_from_seq(&seq)?; - } - } + self.configure(&frame)?; // A keyframe we couldn't configure (no sequence header) is undecodable. if frame.keyframe && !self.catalog.configured() { @@ -259,6 +247,25 @@ impl Import { Ok(()) } + /// Resolve the config and HDR metadata a keyframe carries inline, without publishing it. + pub(crate) fn configure(&mut self, frame: &Frame) -> Result<()> { + if !frame.keyframe { + return Ok(()); + } + let seq = find_sequence_header(&frame.payload); + // A new sequence header starts a coded sequence whose HDR metadata replaces the last + // one's. It goes first, so a keyframe that also configures publishes once. + // The first header after an av1C is the record's own sequence, which sync samples repeat. + let sequence = seq + .as_ref() + .is_some_and(|seq| self.last_seq.as_ref().is_some_and(|last| last != seq)); + self.catalog.hdr(&mut self.track, find_hdr(&frame.payload), sequence)?; + if let Some(seq) = seq { + self.configure_from_seq(&seq)?; + } + Ok(()) + } + /// Publish split frames, resolving the config from the first keyframe's inline /// sequence header and refining the catalog jitter. pub fn decode(&mut self, frames: impl IntoIterator) -> Result<()> { diff --git a/rs/moq-mux/src/codec/h264/import.rs b/rs/moq-mux/src/codec/h264/import.rs index 82c26367dcd..d2f0bc59b1d 100644 --- a/rs/moq-mux/src/codec/h264/import.rs +++ b/rs/moq-mux/src/codec/h264/import.rs @@ -223,19 +223,7 @@ impl Import { /// keyframe's inline SPS and refining the catalog jitter as it goes. fn write_frames(&mut self, frames: impl IntoIterator) -> Result<()> { for frame in frames { - if frame.keyframe { - let hdr = - crate::codec::color::access_unit(&frame.payload, self.length_size, 1, |h| h & 0x1f == NAL_TYPE_SEI); - // avc1 config arrives out-of-band via initialize(); avc3 carries the SPS inline. - let sps = self.length_size.is_none().then(|| find_sps(&frame.payload)).flatten(); - // A new SPS starts a coded sequence whose HDR metadata replaces the last one's. It goes - // first, so a keyframe that also configures publishes once. - let sequence = sps.as_ref().is_some_and(|sps| self.last_sps.as_ref() != Some(sps)); - self.catalog.hdr(&mut self.track, hdr, sequence)?; - if let Some(sps) = sps { - self.configure_from_sps(&sps)?; - } - } + self.configure(&frame)?; if !self.catalog.configured() { // A keyframe we still can't configure is undecodable, so bail @@ -252,6 +240,24 @@ impl Import { Ok(()) } + /// Resolve the config and HDR metadata a keyframe carries inline, without publishing it. + pub(crate) fn configure(&mut self, frame: &Frame) -> Result<()> { + if !frame.keyframe { + return Ok(()); + } + let hdr = crate::codec::color::access_unit(&frame.payload, self.length_size, 1, |h| h & 0x1f == NAL_TYPE_SEI); + // avc1 config arrives out-of-band via initialize(); avc3 carries the SPS inline. + let sps = self.length_size.is_none().then(|| find_sps(&frame.payload)).flatten(); + // A new SPS starts a coded sequence whose HDR metadata replaces the last one's. It goes + // first, so a keyframe that also configures publishes once. + let sequence = sps.as_ref().is_some_and(|sps| self.last_sps.as_ref() != Some(sps)); + self.catalog.hdr(&mut self.track, hdr, sequence)?; + if let Some(sps) = sps { + self.configure_from_sps(&sps)?; + } + Ok(()) + } + /// Publish split frames, resolving the avc3 config from the first keyframe's /// inline SPS and refining the catalog jitter as it goes. pub fn decode(&mut self, frames: impl IntoIterator) -> Result<()> { diff --git a/rs/moq-mux/src/codec/h265/import.rs b/rs/moq-mux/src/codec/h265/import.rs index 1b280b1e14d..1aae391992c 100644 --- a/rs/moq-mux/src/codec/h265/import.rs +++ b/rs/moq-mux/src/codec/h265/import.rs @@ -225,22 +225,7 @@ impl Import { /// keyframe's inline SPS and refining the catalog jitter as it goes. fn write_frames(&mut self, frames: impl IntoIterator) -> Result<()> { for frame in frames { - if frame.keyframe { - let hdr = crate::codec::color::access_unit(&frame.payload, self.length_size, 2, |h| { - nal_unit_type(h) == scuffle_h265::NALUnitType::PrefixSeiNut - }); - // hvc1 config arrives out-of-band via initialize(); hev1 carries the SPS inline. - // Scanning an hvc1 payload as Annex-B would misread a length prefix as a start - // code (a 322-byte NAL prefixes as `00 00 01 42`, an SPS NAL header). - let sps = self.length_size.is_none().then(|| find_sps(&frame.payload)).flatten(); - // A new SPS starts a coded sequence whose HDR metadata replaces the last one's. It goes - // first, so a keyframe that also configures publishes once. - let sequence = sps.as_ref().is_some_and(|sps| self.last_sps.as_ref() != Some(sps)); - self.catalog.hdr(&mut self.track, hdr, sequence)?; - if let Some(sps) = sps { - self.configure_from_sps(&sps)?; - } - } + self.configure(&frame)?; // A keyframe we still can't configure (no SPS) is undecodable. if frame.keyframe && !self.catalog.configured() { @@ -254,6 +239,28 @@ impl Import { Ok(()) } + /// Resolve the config and HDR metadata a keyframe carries inline, without publishing it. + pub(crate) fn configure(&mut self, frame: &Frame) -> Result<()> { + if !frame.keyframe { + return Ok(()); + } + let hdr = crate::codec::color::access_unit(&frame.payload, self.length_size, 2, |h| { + nal_unit_type(h) == scuffle_h265::NALUnitType::PrefixSeiNut + }); + // hvc1 config arrives out-of-band via initialize(); hev1 carries the SPS inline. + // Scanning an hvc1 payload as Annex-B would misread a length prefix as a start + // code (a 322-byte NAL prefixes as `00 00 01 42`, an SPS NAL header). + let sps = self.length_size.is_none().then(|| find_sps(&frame.payload)).flatten(); + // A new SPS starts a coded sequence whose HDR metadata replaces the last one's. It goes + // first, so a keyframe that also configures publishes once. + let sequence = sps.as_ref().is_some_and(|sps| self.last_sps.as_ref() != Some(sps)); + self.catalog.hdr(&mut self.track, hdr, sequence)?; + if let Some(sps) = sps { + self.configure_from_sps(&sps)?; + } + Ok(()) + } + /// Publish split frames, resolving the config from the first keyframe's inline /// SPS and refining the catalog jitter as it goes. pub fn decode(&mut self, frames: impl IntoIterator) -> Result<()> { diff --git a/rs/moq-mux/src/import/track.rs b/rs/moq-mux/src/import/track.rs index b80839a06c2..87098becbce 100644 --- a/rs/moq-mux/src/import/track.rs +++ b/rs/moq-mux/src/import/track.rs @@ -617,32 +617,43 @@ enum GateState { } impl Gate { - /// The access units of `frames` to publish, and whether the first of them resumes the - /// rendition, so the catalog enables it again. - fn admit(&mut self, mut frames: Vec) -> (Vec, bool) { - let resumed = match self.state { - GateState::Open => false, - GateState::Closed => { - frames.clear(); - false - } + /// Split `frames` into the access units dropped and those to publish, and report whether + /// the first published one resumes the rendition, so the catalog enables it again. + fn admit(&mut self, mut frames: Vec) -> (Vec, Vec, bool) { + let (kept, resumed) = match self.state { + GateState::Open => (0, false), + GateState::Closed => (frames.len(), false), GateState::Resuming => match frames.iter().position(|frame| frame.keyframe) { Some(keyframe) => { - frames.drain(..keyframe); self.state = GateState::Open; - true - } - None => { - frames.clear(); - false + (keyframe, true) } + None => (frames.len(), false), }, }; - self.published |= !frames.is_empty(); - (frames, resumed) + let publish = frames.split_off(kept); + self.published |= !publish.is_empty(); + (frames, publish, resumed) } } +/// Publish split `frames` through a [`TrackStream`]'s gate. +/// +/// A dropped keyframe still configures the importer: a stream may carry its config only in +/// band (an AV1 sequence header), and the keyframe that resumes need not repeat it. +macro_rules! publish { + ($gate:expr, $import:expr, $frames:expr) => {{ + let (dropped, frames, resumed) = $gate.admit($frames); + for frame in &dropped { + $import.configure(frame)?; + } + if resumed { + $import.set_enabled(true)?; + } + $import.decode(frames) + }}; +} + impl TrackStream { /// Create an importer that publishes a single codec onto a reserved track. /// @@ -696,22 +707,14 @@ impl TrackStream { ref mut import, } => { import.initialize(data)?; - let (frames, resumed) = self.gate.admit(split.decode(data, None)?); - if resumed { - import.set_enabled(true)?; - } - import.decode(frames)?; + publish!(self.gate, import, split.decode(data, None)?)?; } TrackStreamKind::Hev1 { ref mut split, ref mut import, } => { import.initialize(data)?; - let (frames, resumed) = self.gate.admit(split.decode(data, None)?); - if resumed { - import.set_enabled(true)?; - } - import.decode(frames)?; + publish!(self.gate, import, split.decode(data, None)?)?; } TrackStreamKind::Av01 { ref mut split, @@ -725,11 +728,7 @@ impl TrackStream { } else { split.decode(data, None)? }; - let (frames, resumed) = self.gate.admit(frames); - if resumed { - import.set_enabled(true)?; - } - import.decode(frames)?; + publish!(self.gate, import, frames)?; } } @@ -739,11 +738,7 @@ impl TrackStream { /// Decode a chunk of the byte stream. pub fn decode(&mut self, data: &[u8]) -> Result<()> { stream!(&mut self.kind, split, import => { - let (frames, resumed) = self.gate.admit(split.decode(data, None)?); - if resumed { - import.set_enabled(true)?; - } - import.decode(frames) + publish!(self.gate, import, split.decode(data, None)?) }) } @@ -752,11 +747,7 @@ impl TrackStream { // The splitter is taken out to flush, so its input storage ends with the input. The import // stays, keeping the finished track and its rendition published. stream!(&mut self.kind, split, import => { - let (tail, resumed) = self.gate.admit(std::mem::take(split).flush(None)?); - if resumed { - import.set_enabled(true)?; - } - import.decode(tail)?; + publish!(self.gate, import, std::mem::take(split).flush(None)?)?; import.finish() }) } @@ -768,8 +759,9 @@ impl TrackStream { /// [`rate::Target::Disabled`](crate::rate::Target::Disabled)). Once anything was published, /// that marks a [`discontinuity`](crate::container::Producer::discontinuity). Every access /// unit decoded until it is enabled again is dropped, so the caller may keep writing the - /// stream. Enabled again, it drops access units until the next keyframe and advertises - /// itself enabled with that keyframe. + /// stream, though a dropped keyframe still updates the codec config. Enabled again, it + /// drops access units until the next keyframe and advertises itself enabled with that + /// keyframe. pub fn set_enabled(&mut self, enabled: bool) -> Result<()> { match (enabled, self.gate.state) { (false, GateState::Open) => { @@ -1675,4 +1667,38 @@ mod tests { // access units and the end marker `finish` bounds the last one with. assert_eq!(collect_groups(subscriber).await, vec![2, 1, 4]); } + + /// A keyframe dropped while disabled still configures the importer: an AV1 stream may send + /// its sequence header once, and the keyframe that resumes need not repeat it. + #[tokio::test(start_paused = true)] + async fn track_stream_configures_from_a_dropped_keyframe() { + let (_, with_header) = av1_init_and_keyframe(); + // A temporal delimiter and a key frame, with no sequence header. + let bare = [0x12, 0x00, 0x32, 0x01, 0x10]; + + let (broadcast, catalog) = new_broadcast(); + let consumer = broadcast.consume(); + let request = broadcast.reserve_track("video").unwrap(); + let mut import = TrackStream::video( + request, + catalog.reserve(), + VideoInit::new(VideoFormat::Av01, Vec::new()), + ) + .unwrap(); + + import.set_enabled(false).unwrap(); + // The next temporal delimiter completes the one carrying the sequence header. + import.decode(&with_header).unwrap(); + kio::clock::sleep(Duration::from_millis(33)).await; + import.decode(&bare).unwrap(); + kio::clock::sleep(Duration::from_millis(33)).await; + + import.set_enabled(true).unwrap(); + import.decode(&bare).unwrap(); + import.finish().unwrap(); + + assert!(catalog.snapshot().video.renditions["video"].enabled); + let frame = first_frame(&consumer, "video").await; + assert!(frame.keyframe, "the resumed group opens on a keyframe"); + } } From ac1bc978d82707f9ff279ebf7d65990d060a35eb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 11:34:07 -0800 Subject: [PATCH 7/8] docs: a disabled stream drops what is written to it Co-Authored-By: Claude Opus 5.5 --- doc/lib/cpp/index.md | 2 +- doc/lib/dart/index.md | 2 +- doc/lib/go/index.md | 2 +- doc/lib/kt/index.md | 2 +- doc/lib/py/index.md | 2 +- doc/lib/swift/index.md | 2 +- rs/moq-ffi/src/test.rs | 2 +- 7 files changed, 7 insertions(+), 7 deletions(-) diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index c6e0b9085ee..df32ceac26d 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -149,7 +149,7 @@ ok(publication->announce({})); // each broadcast mints a fresh instance ok(broadcast->close()); // keep the producer alive while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `moq::MediaProducer::kTrack` or `kStream`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(false)` after a `discontinuity()`, and `set_enabled(true)` on the keyframe that resumes it. A `MediaTrackStreamProducer` finds its own frames: `set_enabled(false)` drops what it is written, and after `set_enabled(true)` it resumes at the stream's next keyframe. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `moq::MediaProducer::kTrack` or `kStream`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(false)` after a `discontinuity()`, and `set_enabled(true)` on the keyframe that resumes it. A `MediaTrackStreamProducer` finds its own frames: `set_enabled(false)` drops what is written to it, and after `set_enabled(true)` it resumes at the stream's next keyframe. ## Futures diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 352d2f5859a..918f16ea5df 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -139,7 +139,7 @@ mine.close(); await moq.close(); ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `TrackMoqMediaProducer` or `StreamMoqMediaProducer`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what it is written, and after `setEnabled(true)` it resumes at the stream's next keyframe. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `TrackMoqMediaProducer` or `StreamMoqMediaProducer`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what is written to it, and after `setEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index a80fe99afee..394381fe5a0 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -119,7 +119,7 @@ _ = video.Finish() broadcast.Close() // keep the producer reachable while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `Lower` in its `media.VideoInit` too, a `*media.TrackProducer` or `*media.TrackStreamProducer`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `SetEnabled(false)` after a `Discontinuity()`, and `SetEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `SetEnabled(false)` drops what it is written, and after `SetEnabled(true)` it resumes at the stream's next keyframe. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `Lower` in its `media.VideoInit` too, a `*media.TrackProducer` or `*media.TrackStreamProducer`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `SetEnabled(false)` after a `Discontinuity()`, and `SetEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `SetEnabled(false)` drops what is written to it, and after `SetEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index a56e6a6e952..c0ccf18475b 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -94,7 +94,7 @@ Moq.connect("https://relay.example.com").use { moq -> } ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `uniffi.moq.MoqMediaProducer.Track` or `.Stream`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what it is written, and after `setEnabled(true)` it resumes at the stream's next keyframe. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, as a `uniffi.moq.MoqMediaProducer.Track` or `.Stream`, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what is written to it, and after `setEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 064f9f0171d..2d36fd5d5f7 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -109,7 +109,7 @@ async def main(): asyncio.run(main()) ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `media.TrackProducer.video` or `media.TrackStreamProducer.video` takes `lower` too, of either kind, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(False)` after a `discontinuity()`, and `set_enabled(True)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `set_enabled(False)` drops what it is written, and after `set_enabled(True)` it resumes at the stream's next keyframe. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `media.TrackProducer.video` or `media.TrackStreamProducer.video` takes `lower` too, of either kind, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(False)` after a `discontinuity()`, and `set_enabled(True)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `set_enabled(False)` drops what is written to it, and after `set_enabled(True)` it resumes at the stream's next keyframe. ## Things to know diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 01b98031473..e002fc1725a 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -95,7 +95,7 @@ session.shutdown() try await session.closed() ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `Media.TrackProducer.video` or `Media.TrackStreamProducer.video` takes `lower` too, as a `Media.Producer` of either kind, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what it is written, and after `setEnabled(true)` it resumes at the stream's next keyframe. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `Media.TrackProducer.video` or `Media.TrackStreamProducer.video` takes `lower` too, as a `Media.Producer` of either kind, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. A `TrackStreamProducer` finds its own frames: `setEnabled(false)` drops what is written to it, and after `setEnabled(true)` it resumes at the stream's next keyframe. ## Things to know diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 0d0f913915f..44ab4b95647 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -2954,7 +2954,7 @@ async fn video_rung_sits_below_its_lower_and_disables() { } /// A byte-stream rung above another sits one priority below it, so a grant short of both funds -/// the lower one first. Disabled, it drops what it is written, and enabled again it resumes at +/// the lower one first. Disabled, it drops what is written to it, and enabled again it resumes at /// the next keyframe in the stream. #[tokio::test] async fn stream_rung_is_funded_after_its_lower_and_resumes_at_a_keyframe() { From 0910a5f79e6a62842591cf34a53e97b13528d938 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 12:21:14 -0800 Subject: [PATCH 8/8] fix(mux): a resumed AV1 keyframe carries the sequence header dropped before it Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/codec/av1/import.rs | 43 +++++++++++++++++++++++++++-- rs/moq-mux/src/codec/h264/import.rs | 8 +++++- rs/moq-mux/src/codec/h265/import.rs | 8 +++++- rs/moq-mux/src/import/track.rs | 10 ++++--- 4 files changed, 61 insertions(+), 8 deletions(-) diff --git a/rs/moq-mux/src/codec/av1/import.rs b/rs/moq-mux/src/codec/av1/import.rs index b61d6e549d0..7041586337a 100644 --- a/rs/moq-mux/src/codec/av1/import.rs +++ b/rs/moq-mux/src/codec/av1/import.rs @@ -30,6 +30,9 @@ pub struct Import { track: crate::container::Producer, catalog: crate::codec::video::Catalog, last_seq: Option, + /// A sequence header only a [`dropped`](Self::dropped) keyframe carried, which the next + /// keyframe written takes if it carries none, so a subscriber can still decode it. + owed_seq: bool, } impl Import { @@ -52,6 +55,7 @@ impl Import { track, catalog, last_seq: None, + owed_seq: false, }; if let Some(config) = import.catalog.initial_config() { import.apply_config(config)?; @@ -232,7 +236,7 @@ impl Import { /// Write split frames to the track, resolving the config from the first /// keyframe's inline sequence header and refining the catalog jitter. fn write_frames(&mut self, frames: impl IntoIterator) -> Result<()> { - for frame in frames { + for mut frame in frames { self.configure(&frame)?; // A keyframe we couldn't configure (no sequence header) is undecodable. @@ -240,6 +244,14 @@ impl Import { return Err(Error::MissingSequenceHeader.into()); } + if frame.keyframe + && std::mem::take(&mut self.owed_seq) + && let Some(seq) = &self.last_seq + && find_sequence_header(&frame.payload).is_none() + { + frame.payload = with_sequence_header(&frame.payload, seq)?; + } + // A pre-keyframe delta has no group to anchor it: the producer returns // MissingKeyframe, which a caller joining mid-stream skips. self.track.write(frame)?; @@ -247,8 +259,18 @@ impl Import { Ok(()) } + /// Account for a frame its caller drops rather than writes: a keyframe still configures the + /// rendition, and its sequence header goes out with the next keyframe written. + pub(crate) fn dropped(&mut self, frame: &Frame) -> Result<()> { + self.configure(frame)?; + if frame.keyframe && find_sequence_header(&frame.payload).is_some() { + self.owed_seq = true; + } + Ok(()) + } + /// Resolve the config and HDR metadata a keyframe carries inline, without publishing it. - pub(crate) fn configure(&mut self, frame: &Frame) -> Result<()> { + fn configure(&mut self, frame: &Frame) -> Result<()> { if !frame.keyframe { return Ok(()); } @@ -273,6 +295,23 @@ impl Import { } } +/// `payload` with the sequence header `seq` after its leading temporal delimiter, where a temporal +/// unit carries it. +fn with_sequence_header(payload: &Bytes, seq: &Bytes) -> Result { + let mut buf = payload.clone(); + let first = ObuIterator::new(&mut buf).next().transpose()?; + let delimiter = first.filter(|obu| { + let mut reader = &obu[..]; + ObuHeader::parse(&mut reader).is_ok_and(|header| header.obu_type == ObuType::TemporalDelimiter) + }); + let at = delimiter.map_or(0, |obu| obu.len()); + let mut out = bytes::BytesMut::with_capacity(payload.len() + seq.len()); + out.extend_from_slice(&payload[..at]); + out.extend_from_slice(seq); + out.extend_from_slice(&payload[at..]); + Ok(out.freeze()) +} + fn is_sequence_header(obu: &[u8]) -> bool { let mut reader = obu; ObuHeader::parse(&mut reader) diff --git a/rs/moq-mux/src/codec/h264/import.rs b/rs/moq-mux/src/codec/h264/import.rs index 7bede54e0fa..120844aff48 100644 --- a/rs/moq-mux/src/codec/h264/import.rs +++ b/rs/moq-mux/src/codec/h264/import.rs @@ -232,8 +232,14 @@ impl Import { Ok(()) } + /// Account for a frame its caller drops rather than writes: a keyframe still configures the + /// rendition. The splitter re-injects the SPS and PPS into the next keyframe itself. + pub(crate) fn dropped(&mut self, frame: &Frame) -> Result<()> { + self.configure(frame) + } + /// Resolve the config and HDR metadata a keyframe carries inline, without publishing it. - pub(crate) fn configure(&mut self, frame: &Frame) -> Result<()> { + fn configure(&mut self, frame: &Frame) -> Result<()> { if !frame.keyframe { return Ok(()); } diff --git a/rs/moq-mux/src/codec/h265/import.rs b/rs/moq-mux/src/codec/h265/import.rs index ce1bef287ca..927496ef689 100644 --- a/rs/moq-mux/src/codec/h265/import.rs +++ b/rs/moq-mux/src/codec/h265/import.rs @@ -232,8 +232,14 @@ impl Import { Ok(()) } + /// Account for a frame its caller drops rather than writes: a keyframe still configures the + /// rendition. The splitter re-injects the VPS, SPS, and PPS into the next keyframe itself. + pub(crate) fn dropped(&mut self, frame: &Frame) -> Result<()> { + self.configure(frame) + } + /// Resolve the config and HDR metadata a keyframe carries inline, without publishing it. - pub(crate) fn configure(&mut self, frame: &Frame) -> Result<()> { + fn configure(&mut self, frame: &Frame) -> Result<()> { if !frame.keyframe { return Ok(()); } diff --git a/rs/moq-mux/src/import/track.rs b/rs/moq-mux/src/import/track.rs index 5b8e7b25026..6bad2f2d7e2 100644 --- a/rs/moq-mux/src/import/track.rs +++ b/rs/moq-mux/src/import/track.rs @@ -640,13 +640,13 @@ impl Gate { /// Publish split `frames` through a [`TrackStream`]'s gate. /// -/// A dropped keyframe still configures the importer: a stream may carry its config only in -/// band (an AV1 sequence header), and the keyframe that resumes need not repeat it. +/// The importer still sees what the gate drops: a stream may carry its config only in band (an +/// AV1 sequence header), and the keyframe that resumes need not repeat it. macro_rules! publish { ($gate:expr, $import:expr, $frames:expr) => {{ let (dropped, frames, resumed) = $gate.admit($frames); for frame in &dropped { - $import.configure(frame)?; + $import.dropped(frame)?; } if resumed { $import.set_enabled(true)?; @@ -1831,12 +1831,14 @@ mod tests { import.decode(&bare).unwrap(); kio::clock::sleep(Duration::from_millis(33)).await; + // Finish flushes the last temporal unit, the keyframe that resumes. import.set_enabled(true).unwrap(); - import.decode(&bare).unwrap(); import.finish().unwrap(); assert!(catalog.snapshot().video.renditions["video"].enabled); let frame = first_frame(&consumer, "video").await; assert!(frame.keyframe, "the resumed group opens on a keyframe"); + // The dropped sequence header goes out with it, after its temporal delimiter. + assert_eq!(frame.payload.as_ref(), with_header.as_slice()); } }