fix(mux): split PAT and PMT sections across TS packets

quest/m1/README.md

@@ -162,7 +162,6 @@ session outcomes, and embedder refusals inside session outcomes.
 - [Fail loud on dropped timed metadata](/quest/m1/drop-loud.md) - FLV script tags and fMP4 emsg boxes are counted and warned about instead of silently dropped, until real carriage lands
 - [Encoder colour](/quest/m1/color-model.md) - every moq-video encode path signals the colour its output actually has, or refuses instead of mislabelling
 - [T-STD TS export](/quest/m1/tstd/README.md) - `moq export ts` is a proper remux that passes the T-STD buffer model, starting with a fixed `--delay`
-- [Multi-packet PMT](/quest/m1/ts-pmt-split.md) - `export ts` splits a PMT longer than one packet instead of exiting before its first packet
 - [ATSC AC-3](/quest/m1/ts-atsc-ac3.md) - `export ts` carries 48 kHz ATSC AC-3 at every A/52 rate up to 640 kb/s instead of aborting from 384 kb/s up
 - [Rust non-continuous signal](/quest/m1/rust-continuous.md) - the Rust container consumer reports a frame after a subscribe or discontinuity as non-continuous, like JS, for the tune-in and warmup trims
 - [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - a viewer joining at a recovery point drops the leading pictures it cannot decode; continuous viewers keep them

rs/moq-mux/src/container/ts/export.rs

@@ -515,8 +515,8 @@ impl Kind {
 
 /// The program tables plus the resolved PID layout.
 struct Psi {
-	pat: Pat,
-	pmt: Pmt,
+	pat: Bytes,
+	pmt: Bytes,
 	pcr_pid: u16,
 	/// PID the PMT rides on: the source's original (preserved from the service
 	/// record) or the synthesized [`PMT_PID`] for a media-only source.
@@ -1764,8 +1764,8 @@ impl<E: catalog::Catalog> Export<E> {
 			self.liveness.register(track.pid);
 		}
 		self.psi = Some(Psi {
-			pat,
-			pmt,
+			pat: super::psi::encode_pat(&pat)?,
+			pmt: super::psi::encode_pmt(&pmt)?,
 			pcr_pid,
 			pmt_pid,
 		});
@@ -1839,12 +1839,10 @@ impl<E: catalog::Catalog> Export<E> {
 		if psi {
 			let psi = self.psi.as_ref().context("PSI not built")?;
 			let pmt_pid = psi.pmt_pid;
-			let pat = TsPayload::Pat(psi.pat.clone());
-			let pmt = TsPayload::Pmt(psi.pmt.clone());
-			let flag = self.breaks(Pid::PAT).then(|| flags(true));
-			self.write_packet(&mut out, Pid::PAT, flag, pat)?;
-			let flag = self.breaks(pmt_pid).then(|| flags(true));
-			self.write_packet(&mut out, pmt_pid, flag, pmt)?;
+			let pat = psi.pat.clone();
+			let pmt = psi.pmt.clone();
+			self.write_section(&mut out, Pid::PAT, &pat)?;
+			self.write_section(&mut out, pmt_pid, &pmt)?;
 			self.last_psi = Some(frame.timestamp);
 		}
 
@@ -2143,7 +2141,7 @@ impl<E: catalog::Catalog> Export<E> {
 		Ok(())
 	}
 
-	/// Packetize a private section (SCTE-35 or other) verbatim. The first packet
+	/// Packetize a complete PSI or private section verbatim. The first packet
 	/// carries the pointer_field plus the section start as a `Section` payload (sets
 	/// the unit-start bit so the receiver finds the pointer_field); continuations are
 	/// `Raw`. The section bytes are opaque, so this round-trips byte-for-byte.

rs/moq-mux/src/container/ts/export_test.rs

@@ -7546,6 +7546,123 @@ async fn export_stats_count_a_gap_longer_than_the_backfill() {
 	assert!(quiet < Duration::from_millis(200), "{quiet:?}");
 }
 
+#[tokio::test(start_paused = true)]
+async fn a_large_pmt_is_split_across_packets() {
+	assert_eq!(large_pmt(24, None).await, 2);
+}
+
+/// A PMT spanning three packets at 20 Mb/s goes out at TBsys's rate: the PAT and PMT share its
+/// 512 bytes, which drain at Rxsys (ISO 13818-1 2.4.2.3).
+#[tokio::test(start_paused = true)]
+async fn a_large_pmt_keeps_within_the_system_buffer() {
+	assert_eq!(large_pmt(48, Some(20_000_000)).await, 3);
+}
+
+/// Export `streams` language-tagged AAC streams, check the PMT reassembles across its packets
+/// and keeps within TBsys at `mux_rate`, and return how many packets it spans.
+async fn large_pmt(streams: u16, mux_rate: Option<u64>) -> usize {
+	let mut broadcast = moq_net::broadcast::Info::new().produce();
+	let mut catalog = crate::catalog::Producer::new(
+		&mut broadcast,
+		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
+	)
+	.unwrap();
+	let mut config = tscat::Mpegts::default();
+	let mut producers = Vec::new();
+	for index in 0..streams {
+		let name = format!("audio{index:02}");
+		let track = broadcast
+			.create_track(
+				name.as_str(),
+				hang::container::track_info(hang::catalog::PRIORITY.audio),
+			)
+			.unwrap();
+		catalog
+			.modify()
+			.unwrap()
+			.audio
+			.renditions
+			.insert(name.clone(), AudioConfig::new(AAC { profile: 2 }, 48_000, 2));
+		let mut entry = tscat::Track::new(0x1001 + index);
+		entry.descriptors = vec![tscat::Descriptor {
+			tag: 0x0a,
+			data: Bytes::from_static(b"eng\0"),
+		}];
+		config.tracks.insert(name, entry);
+		let mut producer = Producer::new(track, HangContainer::Legacy(Kind::Data));
+		producer
+			.write(Frame {
+				timestamp: Timestamp::ZERO,
+				duration: None,
+				keyframe: true,
+				payload: Bytes::from_static(b"aac"),
+			})
+			.unwrap();
+		producer.finish().unwrap();
+		producers.push(producer);
+	}
+	catalog.modify().unwrap().ext.mpegts = config;
+	let mut exporter = Export::with_ts(crate::source::announced(&broadcast.consume()).await, Default::default())
+		.await
+		.unwrap();
+	if let Some(mux_rate) = mux_rate {
+		exporter = exporter.with_mux_rate(mux_rate);
+	}
+	let ts = drain_with(exporter).await;
+	let packets = ts.as_chunks::<188>().0;
+	let pid = |packet: &[u8; 188]| (u16::from(packet[1] & 0x1f) << 8) | u16::from(packet[2]);
+	if let Some(mux_rate) = mux_rate {
+		// TBsys: 188 B in per table packet, drained at Rxsys over each packet's time.
+		let drain = 1_000_000.0 / mux_rate as f64 * 188.0;
+		let (mut tb, mut peak) = (0.0f64, 0.0f64);
+		for packet in packets {
+			tb = (tb - drain).max(0.0);
+			if [0, 0x1000].contains(&pid(packet)) {
+				tb += 188.0;
+				peak = peak.max(tb);
+			}
+		}
+		assert!(peak <= 512.0, "TBsys peak {peak:.0} B");
+	}
+	let mut section = Vec::new();
+	let mut counters = Vec::new();
+	for packet in packets {
+		if pid(packet) != 0x1000 {
+			continue;
+		}
+		counters.push(packet[3] & 0x0f);
+		let offset = if packet[3] & 0x20 != 0 {
+			5 + usize::from(packet[4])
+		} else {
+			4
+		};
+		if packet[1] & 0x40 != 0 {
+			assert!(section.is_empty());
+			assert_eq!(packet[offset], 0);
+			section.extend_from_slice(&packet[offset + 1..]);
+		} else {
+			section.extend_from_slice(&packet[offset..]);
+		}
+		let length = 3 + (((usize::from(section[1]) & 0x0f) << 8) | usize::from(section[2]));
+		if section.len() >= length {
+			section.truncate(length);
+			break;
+		}
+	}
+	for pair in counters.windows(2) {
+		assert_eq!(pair[1], (pair[0] + 1) & 0x0f);
+	}
+	assert!(super::psi::crc_ok(&section));
+	let pmt = super::psi::Pmt::parse(&section).unwrap();
+	assert_eq!(pmt.streams.len(), usize::from(streams));
+	for (index, stream) in pmt.streams.iter().enumerate() {
+		assert_eq!(stream.pid.as_u16(), 0x1001 + index as u16);
+		assert_eq!(stream.stream_type, 0x0f);
+		assert_eq!(stream.descriptors[0].data.as_ref(), b"eng\0");
+	}
+	counters.len()
+}
+
 /// Publish a catalog-only broadcast at `live`, for a test of what follows its end.
 fn publish_bare(
 	origin: &moq_net::origin::Producer,

rs/moq-mux/src/container/ts/psi.rs

@@ -24,6 +24,58 @@ pub(super) fn crc_ok(section: &[u8]) -> bool {
 	CRC.checksum(body) == u32::from_be_bytes(crc.try_into().expect("four bytes"))
 }
 
+/// Encode the exporter's single-program PAT independently of TS packet boundaries.
+pub(super) fn encode_pat(pat: &mpeg2ts::ts::payload::Pat) -> anyhow::Result<bytes::Bytes> {
+	let mut section = vec![PAT_TABLE_ID, 0xb0, 0];
+	section.extend_from_slice(&pat.transport_stream_id.to_be_bytes());
+	section.extend_from_slice(&[0xc1 | (pat.version_number.as_u8() << 1), 0, 0]);
+	for entry in &pat.table {
+		section.extend_from_slice(&entry.program_num.to_be_bytes());
+		section.extend_from_slice(&(0xe000 | entry.program_map_pid.as_u16()).to_be_bytes());
+	}
+	finish_section(section)
+}
+
+/// Encode a PMT, rejecting descriptor or table lengths that MPEG-TS cannot represent.
+pub(super) fn encode_pmt(pmt: &mpeg2ts::ts::payload::Pmt) -> anyhow::Result<bytes::Bytes> {
+	let mut section = vec![PMT_TABLE_ID, 0xb0, 0];
+	section.extend_from_slice(&pmt.program_num.to_be_bytes());
+	section.extend_from_slice(&[0xc1 | (pmt.version_number.as_u8() << 1), 0, 0]);
+	section.extend_from_slice(&(0xe000 | pmt.pcr_pid.map_or(0x1fff, |pid| pid.as_u16())).to_be_bytes());
+	encode_descriptors(&mut section, &pmt.program_info)?;
+	for stream in &pmt.es_info {
+		section.push(stream.stream_type as u8);
+		section.extend_from_slice(&(0xe000 | stream.elementary_pid.as_u16()).to_be_bytes());
+		encode_descriptors(&mut section, &stream.descriptors)?;
+	}
+	finish_section(section)
+}
+
+fn encode_descriptors(out: &mut Vec<u8>, descriptors: &[mpeg2ts::ts::Descriptor]) -> anyhow::Result<()> {
+	let length = descriptors
+		.iter()
+		.map(|descriptor| 2 + descriptor.data.len())
+		.sum::<usize>();
+	anyhow::ensure!(length <= 0x0fff, "PSI descriptor loop exceeds 4095 bytes");
+	out.extend_from_slice(&(0xf000 | length as u16).to_be_bytes());
+	for descriptor in descriptors {
+		let length = u8::try_from(descriptor.data.len()).context("PSI descriptor exceeds 255 bytes")?;
+		out.extend_from_slice(&[descriptor.tag, length]);
+		out.extend_from_slice(&descriptor.data);
+	}
+	Ok(())
+}
+
+fn finish_section(mut section: Vec<u8>) -> anyhow::Result<bytes::Bytes> {
+	let length = section.len() - 3 + 4;
+	anyhow::ensure!(length <= 1021, "PAT/PMT section exceeds 1021 bytes");
+	section[1] |= (length >> 8) as u8;
+	section[2] = length as u8;
+	let crc = CRC.checksum(&section);
+	section.extend_from_slice(&crc.to_be_bytes());
+	Ok(section.into())
+}
+
 /// The table_id of a PAT section.
 pub(super) const PAT_TABLE_ID: u8 = 0x00;
 /// The table_id of a PMT section.
@@ -315,6 +367,38 @@ pub(super) fn section(table_id: u8, extension: u16, version: u8, number: u8, las
 mod test {
 	use super::*;
 
+	#[test]
+	fn pmt_encoding_rejects_unrepresentable_lengths() {
+		use mpeg2ts::ts::payload::Pmt;
+		use mpeg2ts::ts::{Descriptor, VersionNumber};
+		let mut pmt = Pmt {
+			program_num: 1,
+			pcr_pid: None,
+			version_number: VersionNumber::default(),
+			program_info: vec![Descriptor {
+				tag: 0x80,
+				data: vec![0; 256],
+			}],
+			es_info: Vec::new(),
+		};
+		assert!(encode_pmt(&pmt).unwrap_err().to_string().contains("255 bytes"));
+		pmt.program_info = vec![
+			Descriptor {
+				tag: 0x80,
+				data: vec![0; 250]
+			};
+			4
+		];
+		let section = encode_pmt(&pmt).unwrap();
+		assert!(crc_ok(&section));
+		assert_eq!(section.len(), 1024);
+		pmt.program_info.push(Descriptor {
+			tag: 0x81,
+			data: Vec::new(),
+		});
+		assert!(encode_pmt(&pmt).unwrap_err().to_string().contains("1021 bytes"));
+	}
+
 	#[test]
 	fn crc_matches_mpeg2() {
 		// The single-program PAT ffmpeg writes: TSID 1, program 1 on PID 0x1000.