diff --git a/drafts/draft-lcurley-moq-mpegts.md b/drafts/draft-lcurley-moq-mpegts.md index 2a11af3a0e..02c2018ab6 100644 --- a/drafts/draft-lcurley-moq-mpegts.md +++ b/drafts/draft-lcurley-moq-mpegts.md @@ -169,6 +169,7 @@ type SiEntry = { JSON object keys are strings, so both are decimal with no leading zeros: `"17"` for PID 0x0011, `"66"` for `table_id` 0x42. A consumer MUST refuse a catalog whose PID key is not an integer in 0..8191, or whose `table_id` key is not an integer in 0..255. +A consumer MAY additionally accept the pre-`table_id` form, where a PID maps to `{"interval", "sections"}` with the sections inline: it decodes into one entry per `table_id` (byte 0 of each section) naming no track. A producer MUST NOT write that form. `table_id` is byte 0 of generic section syntax ({{mpeg2}} Section 2.4.4), so the key is no less generic than the PID; which ranges mean what is a delivery-system convention this document does not rely on. @@ -344,6 +345,7 @@ A broadcast demultiplexed from a DVB transport stream: video and audio described - The `Si` type is keyed by `table_id` only; the PID lives on the enclosing `si` map. - A consumer refuses a catalog with an unrecognized `framing` or an invalid `si` map key. - Added `muxRate`, the source's constant multiplex rate. +- A consumer may read the pre-`table_id` inline `sections` form; writing stays track-only. # Acknowledgments diff --git a/rs/moq-mux/src/container/ts/catalog.rs b/rs/moq-mux/src/container/ts/catalog.rs index 1446eb09c0..78449bb0bc 100644 --- a/rs/moq-mux/src/container/ts/catalog.rs +++ b/rs/moq-mux/src/container/ts/catalog.rs @@ -21,7 +21,7 @@ use std::time::Duration; use bytes::Bytes; use serde::{Deserialize, Serialize}; use serde_with::base64::Base64; -use serde_with::{DisplayFromStr, DurationMilliSeconds, serde_as}; +use serde_with::{DurationMilliSeconds, serde_as}; use crate::catalog::hang::CatalogExt; @@ -115,8 +115,19 @@ pub struct Mpegts { /// JSON object keys are strings, so both keys are written in decimal (`"17"`) /// rather than as numbers. The catalog is parsed via `serde_json::Value`, which /// will not coerce a string key back to an integer on its own. - #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] - #[serde_as(as = "BTreeMap>")] + /// + /// Reading also accepts the form every published moq-cli through 0.11 wrote, + /// `{"17": {"interval": 2000, "sections": [""]}}`: the sections inline + /// under the PID with no `table_id` level. Those decode into one entry per + /// `table_id` (byte 0 of each section) carrying its sections in + /// `SiEntry::sections` and naming no track. Nothing writes that form: + /// serializing such an entry fails rather than emit a dangling track reference. + #[serde( + default, + skip_serializing_if = "BTreeMap::is_empty", + serialize_with = "serialize_si", + deserialize_with = "deserialize_si" + )] pub si: BTreeMap>, /// The rate the PCR clock paces the whole multiplex at, in bits per second: @@ -189,6 +200,93 @@ pub struct SiEntry { #[serde_as(as = "Option>")] #[serde(default, skip_serializing_if = "Option::is_none")] pub interval: Option, + + /// The sections themselves, only for an entry read from the pre-`table_id` + /// catalog form (see [`Mpegts::si`]): that form carried them inline, so there + /// is no track to subscribe to and export re-emits these instead. Empty for an + /// entry that names a track. Serializing an entry that carries them fails. + #[serde(skip)] + pub(crate) sections: Vec, +} + +/// Encode [`Mpegts::si`] with both integer keys as decimal strings. +/// +/// An entry carrying inline sections (see [`Mpegts::si`]) names no track, so +/// there is nothing faithful to write for it: serialization fails rather than +/// emit a dangling empty track reference. +fn serialize_si( + si: &BTreeMap>, + serializer: S, +) -> Result { + use serde::ser::{Error, SerializeMap}; + let mut map = serializer.serialize_map(Some(si.len()))?; + for (pid, tables) in si { + for entry in tables.values() { + if !entry.sections.is_empty() { + return Err(S::Error::custom(format!( + "inline SI sections on PID {pid} name no track and cannot be serialized" + ))); + } + } + let tables: BTreeMap = tables.iter().map(|(id, entry)| (id.to_string(), entry)).collect(); + map.serialize_entry(&pid.to_string(), &tables)?; + } + map.end() +} + +/// Decode [`Mpegts::si`] in either form: the `table_id`-keyed map of track +/// references, or the pre-`table_id` inline form under the same PID keys. +fn deserialize_si<'de, D>(deserializer: D) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, +{ + use serde::de::Error; + + /// The pre-`table_id` form of one PID's entry. + #[serde_as] + #[derive(Deserialize)] + #[serde(rename_all = "camelCase", deny_unknown_fields)] + struct Inline { + #[serde_as(as = "Vec")] + sections: Vec, + #[serde_as(as = "Option>")] + #[serde(default)] + interval: Option, + } + + let raw: BTreeMap = Deserialize::deserialize(deserializer)?; + let mut si = BTreeMap::new(); + for (pid, value) in raw { + let pid: u16 = pid.parse().map_err(D::Error::custom)?; + let tables = if value.get("sections").is_some() { + let inline: Inline = serde_json::from_value(value).map_err(D::Error::custom)?; + let mut tables: BTreeMap = BTreeMap::new(); + for section in inline.sections { + let Some(&table_id) = section.first() else { + return Err(D::Error::custom(format!("empty SI section on PID {pid}"))); + }; + tables + .entry(table_id) + .or_insert_with(|| SiEntry { + // The inline form's interval covered the whole PID; a table it + // left unbounded gets the DVB maximum export would use anyway. + interval: inline.interval.or_else(|| si_interval(table_id)), + ..Default::default() + }) + .sections + .push(section); + } + tables + } else { + let tables: BTreeMap = serde_json::from_value(value).map_err(D::Error::custom)?; + tables + .into_iter() + .map(|(table_id, entry)| Ok((table_id.parse().map_err(D::Error::custom)?, entry))) + .collect::>()? + }; + si.insert(pid, tables); + } + Ok(si) } /// One track's MPEG-TS identity and signaling. @@ -433,6 +531,66 @@ mod test { assert_eq!(parsed.mpegts, mpegts, "program and SI round-trip"); } + #[test] + fn inline_si_form_is_read() { + // The `mpegts` section moq-cli 0.11.0 publishes for an ffmpeg TS with an SDT: + // the sections inline under the PID, no `table_id` level. Every published CLI + // through 0.11 writes this, so a consumer that refuses it drops SRT/TS export + // for every one of them. + let json = r#"{ + "program": {"pmtPid": 4096, "programNumber": 1, "transportStreamId": 1}, + "si": { + "17": { + "interval": 2000, + "sections": ["QvAlAAHBAAD/Af8AAfyAFEgSAQZGRm1wZWcJU2VydmljZTAxd3xDyg=="] + } + }, + "tracks": {"0.avc3": {"pid": 256}} + }"#; + let mpegts: Mpegts = serde_json::from_str(json).expect("the 0.11.0 form must decode"); + + // The one SDT section keys itself by its table_id (0x42), carries its bytes + // inline, keeps the PID's interval, and names no track. + let sdt = &mpegts.si[&0x0011][&0x42]; + assert_eq!(sdt.track, ""); + assert_eq!(sdt.interval, Some(Duration::from_secs(2))); + assert_eq!(sdt.sections.len(), 1); + assert_eq!(sdt.sections[0][0], 0x42); + assert_eq!(mpegts.si[&0x0011].len(), 1); + + // Two tables on one PID split into two entries, and a PID without an + // interval falls back to the DVB maximum for each table it carries. + let json = r#"{"si": {"18": {"sections": ["TgAB", "UAAB", "TgAC"]}}}"#; + let mpegts: Mpegts = serde_json::from_str(json).unwrap(); + let eit = &mpegts.si[&0x0012]; + assert_eq!(eit.len(), 2); + assert_eq!(eit[&0x4e].sections.len(), 2); + assert_eq!(eit[&0x4e].interval, Some(Duration::from_secs(2))); + assert_eq!(eit[&0x50].sections.len(), 1); + assert_eq!(eit[&0x50].interval, Some(Duration::from_secs(10))); + + // The inline form never comes back out: an entry carrying sections names + // no track, so serialization fails rather than emit a dangling reference. + serde_json::to_string(&mpegts).expect_err("inline sections must not serialize"); + } + + #[test] + fn track_backed_si_roundtrip_after_legacy_read() { + // A catalog mixing the legacy inline form with a track-backed entry still + // writes the track-backed half; only the entry naming no track is refused. + let json = r#"{"si": { + "17": {"sections": ["QvAlAAHBAAD/Af8AAfyAFEgSAQZGRm1wZWcJU2VydmljZTAxd3xDyg=="]}, + "18": {"78": {"track": "si/18/78", "interval": 2000}} + }}"#; + let mut mpegts: Mpegts = serde_json::from_str(json).unwrap(); + serde_json::to_string(&mpegts).expect_err("the inline entry must not serialize"); + mpegts.si.remove(&0x0011); + let json = serde_json::to_string(&mpegts).unwrap(); + assert!(json.contains("\"si/18/78\""), "track-backed entry still writes: {json}"); + let parsed: Mpegts = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed, mpegts); + } + #[test] fn unknown_framing_is_refused() { let json = r#"{ @@ -456,6 +614,17 @@ mod test { serde_json::from_str::(json).expect_err("a non-integer SI PID key must fail"); } + #[test] + fn mixed_si_forms_are_refused() { + // `sections` alongside any other key would silently drop the new-form + // entries: the inline reader ignores unknown keys without this. + let json = r#"{ "si": { "17": { + "sections": ["QvAlAAHBAAD/Af8AAfyAFEgSAQZGRm1wZWcJU2VydmljZTAxd3xDyg=="], + "66": { "track": "si/17/66" } + } } }"#; + serde_json::from_str::(json).expect_err("sections with a table_id entry must fail"); + } + /// The `mpegts` section is not hang-only: the same JSON rides the MSF catalog track, so a /// broadcast demuxed from MPEG-TS can be re-muxed from either catalog. #[tokio::test] diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index c162ca7ee9..ae9e52ff62 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -333,12 +333,23 @@ enum SiState { } impl SiTrack { - fn new(source: &crate::Source, track: &str, interval: Option, max_age: Duration) -> Self { + fn new(source: &crate::Source, entry: &catalog::SiEntry, max_age: Duration) -> Self { + // An entry read from the inline catalog form has its sections in hand and + // no track to subscribe to: it starts (and stays) where an ended track ends + // up, re-emitting a fixed snapshot on its interval. + let (state, active) = if entry.sections.is_empty() { + ( + SiState::Requesting(source.request_catalog(), entry.track.clone()), + Default::default(), + ) + } else { + (SiState::Done, Self::inline(entry)) + }; Self { - track: track.to_string(), - interval, - state: SiState::Requesting(source.request_catalog(), track.to_string()), - active: Default::default(), + track: entry.track.clone(), + interval: entry.interval, + state, + active, pending: None, dirty: false, last_emit: None, @@ -346,6 +357,15 @@ impl SiTrack { } } + /// The snapshot an inline-form entry's sections reduce to. + fn inline(entry: &catalog::SiEntry) -> super::si::Snapshot { + let mut snapshot = super::si::Snapshot::default(); + for section in &entry.sections { + snapshot.apply(section); + } + snapshot + } + /// Drive the subscription and fold arrived groups into `active`. Never returns /// an error: SI is auxiliary, so a failed or ended track logs and keeps the last /// snapshot rather than killing the mux. @@ -813,18 +833,33 @@ impl Export { // staying attached would repeat its stale sections forever. The last // snapshot carries across so emission never goes dark mid-swap. Some(existing) if existing.track != entry.track => { - let mut replacement = SiTrack::new(&self.source, &entry.track, entry.interval, self.max_age); - replacement.active = std::mem::take(&mut existing.active); - replacement.dirty = existing.dirty; + let mut replacement = SiTrack::new(&self.source, entry, self.max_age); + // An inline entry already holds its snapshot; only a track entry + // has nothing to emit until its first group lands. + if replacement.active.is_empty() { + replacement.active = std::mem::take(&mut existing.active); + replacement.dirty = existing.dirty; + } else { + replacement.dirty = replacement.active != existing.active; + } replacement.last_emit = existing.last_emit; *existing = replacement; } - Some(existing) => existing.interval = entry.interval, + Some(existing) => { + existing.interval = entry.interval; + // Inline sections arrive with the catalog itself, so a revised + // table shows up here rather than on a track. + if !entry.sections.is_empty() { + let snapshot = SiTrack::inline(entry); + if existing.active != snapshot { + existing.active = snapshot; + existing.dirty = true; + } + } + } None => { - self.si.insert( - (*pid, *table_id), - SiTrack::new(&self.source, &entry.track, entry.interval, self.max_age), - ); + self.si + .insert((*pid, *table_id), SiTrack::new(&self.source, entry, self.max_age)); } } } diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 2ad0de6deb..aaa062261e 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -2130,6 +2130,95 @@ async fn si_pids_are_re_emitted_on_their_own_interval() { assert_eq!(count(0x0010), 2, "NIT re-emitted on its 10s interval"); } +/// The catalog form every published moq-cli through 0.11 writes carries the SI +/// sections inline under the PID, with no snapshot track. Export must carry them +/// the same way it carries a track's snapshot, on the PID's interval, or every one +/// of those publishers loses its service layer (and, before the inline form was +/// read at all, the whole export). +#[tokio::test(start_paused = true)] +async fn inline_si_form_is_re_emitted() { + use base64::Engine; + + let broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + + let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap(); + let track = broadcast + .create_track( + broadcast.unique_name(".avc1"), + hang::container::track_info(hang::catalog::PRIORITY.video), + ) + .unwrap(); + let name = track.name().to_string(); + + // Hand-written rather than produced, since nothing writes this form any more. + let mut catalog = crate::catalog::hang::Catalog::::default(); + let mut cfg = VideoConfig::new(H264 { + profile: 0x64, + constraints: 0, + level: 0x1f, + inline: false, + }); + cfg.container = Container::Legacy; + cfg.description = Some(avcc); + catalog.video.renditions.insert(name.clone(), cfg); + let mut json = serde_json::to_value(&catalog).unwrap(); + let sdt = make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 8]); + let inline = base64::engine::general_purpose::STANDARD.encode(&sdt); + json["mpegts"] = serde_json::json!({"si": {"17": {"interval": 2000, "sections": [inline]}}}); + // Held open: dropping the producer ends the track before export subscribes. + let mut catalog_track = broadcast + .create_track(hang::Catalog::DEFAULT_NAME, hang::Catalog::default_track_info()) + .unwrap(); + catalog_track + .write_frame(Timestamp::ZERO, Bytes::from(serde_json::to_vec(&json).unwrap())) + .unwrap(); + + // One keyframe per second across 12s, as in the snapshot-track test above. + let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data)); + let mut idr = vec![0x65u8]; + idr.extend(std::iter::repeat_n(0xAB, 300)); + for sec in 0..=12u64 { + producer + .write(Frame { + timestamp: Timestamp::from_micros(sec * 1_000_000).unwrap(), + duration: None, + payload: length_prefixed(&[&idr]), + keyframe: true, + }) + .unwrap(); + producer.cut(None).unwrap(); + } + producer.finish().unwrap(); + + let ts = drain_with( + Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang) + .await + .unwrap(), + ) + .await; + assert_packet_aligned(&ts); + + let count = |pid: u16| { + ts.chunks_exact(188) + .filter(|p| ((((p[1] & 0x1f) as u16) << 8) | p[2] as u16) == pid) + .count() + }; + assert_eq!(count(0x0000), 13, "PAT on every frame"); + // SDT at 0,2,4,6,8,10,12s, byte-for-byte the inline section. + assert_eq!(count(0x0011), 7, "inline SDT re-emitted on its 2s interval"); + let packet = ts + .chunks_exact(188) + .find(|p| ((((p[1] & 0x1f) as u16) << 8) | p[2] as u16) == 0x0011) + .unwrap(); + // The section is stuffed to the packet's tail, byte-for-byte the inline one. + assert_eq!( + &packet[188 - sdt.len()..], + &sdt[..], + "the inline SDT rides its PID: {packet:02x?}" + ); +} + /// Count payload-bearing TS packets on `pid`, excluding its standalone clock packets. fn count_pid(frames: &[Frame], pid: u16) -> usize { frames