Skip to content

fix(mux): keep the last valid parameter sets when refusing a TS video unit #5152

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
kixelated wants to merge 1 commit into from
Closed

fix(mux): keep the last valid parameter sets when refusing a TS video unit #5152

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions rs/moq-mux/src/codec/h264/split.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,12 @@ pub struct Split {
pending: Vec<crate::container::Frame>,
}

/// A snapshot of [`Split`]'s retained parameter sets.
pub(crate) struct Params {
sps: Vec<Bytes>,
pps: Vec<Bytes>,
}

#[derive(Default)]
struct Avc3Frame {
chunks: BytesMut,
Expand Down Expand Up @@ -236,6 +242,21 @@ impl Split {
self.pending.clear();
}

/// The retained parameter sets, to [`restore`](Self::restore) if the caller refuses
/// the access unit that replaced them.
pub(crate) fn params(&self) -> Params {
Params {
sps: self.sps.clone(),
pps: self.pps.clone(),
}
}

/// Put back parameter sets taken by [`params`](Self::params).
pub(crate) fn restore(&mut self, params: Params) {
self.sps = params.sps;
self.pps = params.pps;
}

fn pts(&mut self, hint: Option<moq_net::Timestamp>) -> Result<moq_net::Timestamp> {
if let Some(pts) = hint {
return Ok(pts);
Expand Down
24 changes: 24 additions & 0 deletions rs/moq-mux/src/codec/h265/split.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,13 @@ pub struct Split {
pending: Vec<crate::container::Frame>,
}

/// A snapshot of [`Split`]'s retained parameter sets.
pub(crate) struct Params {
vps: Vec<Bytes>,
sps: Vec<Bytes>,
pps: Vec<Bytes>,
}

#[derive(Default)]
struct Au {
chunks: BytesMut,
Expand Down Expand Up @@ -240,6 +247,23 @@ impl Split {
self.pending.clear();
}

/// The retained parameter sets, to [`restore`](Self::restore) if the caller refuses
/// the access unit that replaced them.
pub(crate) fn params(&self) -> Params {
Params {
vps: self.vps.clone(),
sps: self.sps.clone(),
pps: self.pps.clone(),
}
}

/// Put back parameter sets taken by [`params`](Self::params).
pub(crate) fn restore(&mut self, params: Params) {
self.vps = params.vps;
self.sps = params.sps;
self.pps = params.pps;
}

fn pts(&mut self, hint: Option<moq_net::Timestamp>) -> Result<moq_net::Timestamp> {
if let Some(pts) = hint {
return Ok(pts);
Expand Down
119 changes: 105 additions & 14 deletions rs/moq-mux/src/container/ts/import.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1849,13 +1849,24 @@ impl<E: catalog::Catalog> Stream<E> {
Stream::H264 { split, import, unwrap } => {
let reorder = reorder_delay(pending.pts, pending.dts);
let pts = unwrap_pts(unwrap, pending.pts, pending.offset)?;
// Each PES is one access unit, so flush to emit it immediately.
let mut frames = split.decode(&pending.data, pts).map_err(unit_error)?;
frames.extend(split.flush(pts).map_err(unit_error)?);
let mut published = 0;
for frame in frames {
published += u64::from(skip_missing_keyframe(import.decode([frame]))?);
}
// A refused unit must not leave its parameter sets behind for later bare
// keyframes to re-inject, or every one of them is refused too.
let params = split.params();

@chatgpt-codex-connector chatgpt-codex-connector Bot Oct 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Avoid cloning parameter sets for every video PES

After the first keyframe, every normal H.264 PES now clones the retained SPS and PPS vectors, and the equivalent H.265 path performs three clones, even though delta access units cannot replace those caches and damage is exceptional. Because cloning each nonempty Vec<Bytes> allocates, steady-state video ingestion gains two or three heap allocations per frame; snapshot only when the unit can update parameter sets, or validate this hot-path cost with a benchmark and use an allocation-free rollback mechanism.

AGENTS.md reference: AGENTS.md:L53-L53

Useful? React with 👍 / 👎.

let published = (|| {
// Each PES is one access unit, so flush to emit it immediately.
let mut frames = split.decode(&pending.data, pts).map_err(unit_error)?;
frames.extend(split.flush(pts).map_err(unit_error)?);
let mut published = 0;
for frame in frames {
published += u64::from(skip_missing_keyframe(import.decode([frame]))?);
}
anyhow::Ok(published)
})()
.inspect_err(|err| {
if err.is::<Damaged>() {
split.restore(params);
}
})?;
// After decode, so the track (and its catalog rendition) exists.
if let Some(reorder) = reorder {
import.observe_reorder(reorder)?;
Expand All @@ -1865,13 +1876,23 @@ impl<E: catalog::Catalog> Stream<E> {
Stream::H265 { split, import, unwrap } => {
let reorder = reorder_delay(pending.pts, pending.dts);
let pts = unwrap_pts(unwrap, pending.pts, pending.offset)?;
// Each PES is one access unit, so flush to emit it immediately.
let mut frames = split.decode(&pending.data, pts).map_err(unit_error)?;
frames.extend(split.flush(pts).map_err(unit_error)?);
let mut published = 0;
for frame in frames {
published += u64::from(skip_missing_keyframe(import.decode([frame]))?);
}
// See the H.264 arm.
let params = split.params();
let published = (|| {
// Each PES is one access unit, so flush to emit it immediately.
let mut frames = split.decode(&pending.data, pts).map_err(unit_error)?;
frames.extend(split.flush(pts).map_err(unit_error)?);
let mut published = 0;
for frame in frames {
published += u64::from(skip_missing_keyframe(import.decode([frame]))?);
}
anyhow::Ok(published)
})()
.inspect_err(|err| {
if err.is::<Damaged>() {
split.restore(params);
}
})?;
if let Some(reorder) = reorder {
import.observe_reorder(reorder)?;
}
Expand Down Expand Up @@ -6732,6 +6753,76 @@ pub(super) mod test {
damaged_video_recovers("clean").await;
}

/// A keyframe refused for its malformed SPS leaves the last valid parameter sets for the
/// next bare keyframe to re-inject. Kept, the malformed SPS would refuse every one of them.
async fn refused_params_keep_the_last_valid(
stream_type: StreamType,
params: &[&[u8]],
damaged: &[&[u8]],
idr: &[u8],
delta: &[u8],
) {
let au = |nals: &[&[u8]]| {
let mut out = Vec::new();
for nal in nals {
out.extend_from_slice(&[0, 0, 0, 1]);
out.extend_from_slice(nal);
}
out
};
let units = [
au(&[params, &[idr]].concat()),
au(&[delta]),
au(&[damaged, &[idr]].concat()),
au(&[idr]),
au(&[delta]),
];
let mut data = synth_pmt(&[(stream_type, VIDEO)], false);
for (i, unit) in units.iter().enumerate() {
let pts = 90_000 + i as u64 * FRAME;
data.extend_from_slice(&video_pes(VIDEO, i as u8, pts, None, unit));
}

let (consumer, catalog, import) = import_all(&data).expect("damage must stay local to the unit");
assert_eq!(import.stats().streams[&VIDEO].damaged, 1);
let name = catalog.snapshot().video.renditions.keys().next().unwrap().clone();
let frames = read_track(&consumer, &name, crate::container::Kind::Video).await;
assert_eq!(frames.len(), 4, "only the damaged keyframe is refused");
assert!(frames[2].keyframe, "the bare keyframe recovers");
let mut expected = au(params);
expected.extend_from_slice(&au(&[idr]));
assert_eq!(
frames[2].payload, expected,
"the last valid parameter sets are re-injected"
);
}

#[tokio::test(start_paused = true)]
async fn refused_h264_params_keep_the_last_valid() {
use crate::container::test_util::{IDR, PPS, SPS};
refused_params_keep_the_last_valid(
StreamType::H264,
&[SPS, PPS],
&[&[0x67], PPS],
IDR,
&[0x41, 0x9a, 0x00, 0x01],
)
.await;
}

#[tokio::test(start_paused = true)]
async fn refused_h265_params_keep_the_last_valid() {
use crate::codec::h265::fixtures::{PPS, SPS, VPS};
refused_params_keep_the_last_valid(
StreamType::H265,
&[VPS, SPS, PPS],
&[VPS, &SPS[..8], PPS],
&[0x26, 0x01, 0x80, 0xaa],
&[0x02, 0x01, 0x80, 0x55],
)
.await;
}

/// A dedicated PCR PID's malformed adaptation field cannot declare a timebase break.
#[test]
fn damaged_adaptation_on_a_dedicated_pcr_pid_keeps_the_clock() {
Expand Down
Loading