fix(ts): schedule padded output at exact decoder times

.github/workflows/nightly.yml

@@ -219,6 +219,10 @@ jobs:
         if: ${{ !cancelled() }}
         run: nix develop --command just rs test -p moq-mux benchmark_damage_pid_fanout --run-ignored only --success-output immediate
 
+      - name: TS scheduler benchmark smoke
+        if: ${{ !cancelled() }}
+        run: nix develop --command just rs bench-ts-schedule --test
+
       - name: Container export benchmark smoke
         if: ${{ !cancelled() }}
         run: nix develop --command just rs bench-export --test

quest/m1/README.md

@@ -144,7 +144,6 @@ above them, since the epoch anchor requires it.
 - [Encoder colour](/quest/m1/color-model.md) - every moq-video encode path signals the colour its output actually has, or refuses instead of mislabelling
 - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide
 - [T-STD TS export](/quest/m1/tstd/README.md) - `moq export ts` is a proper remux that passes the supported T-STD buffer models, starting with a fixed `--delay`
-- [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
 - [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
 - [Audio warmup](/quest/m1/audio-warmup.md) - a viewer joining an Opus rendition mid-stream never hears the unconverged first 80 ms
 - [#3021](/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md) - moq-gst picks the broadcast wall epoch; a restarted source is a new epoch, not a forward re-anchor

quest/m1/ts-live-clock.md

@@ -56,4 +56,3 @@ public API and wire changes are not expected.
 - [Same-instance resume](/quest/m1/ts-same-instance-resume.md) - independent full-check failure
 - [Transport upgrade](/quest/m1/transport-upgrade/README.md) - existing session handover and epoch contract
 - [Mux-rate hold #5287](https://github.com/moq-dev/moq/pull/5287) - initial rate publication, and a finished import lingering for its max-age
-- [TS AC-3 scheduling](/quest/m1/ts-atsc-ac3.md) - #5306 owns decoder deadlines and unknown-rate startup overflow

quest/m1/ts-reliability/README.md

@@ -49,9 +49,9 @@ and the MPEG-TS draft consistent in each feature change.
 The planning children settle catalog and clock contracts before production
 changes. Their output adds independently completable implementation quests to
 this line; an audit finding is not authorization to guess a public API.
-PMT fragmentation (#5220) and initial mux-rate measurement (#5287) have
-landed; PSI runs, decoder-time admission, and catalog linger remain with
-their current quests.
+PMT fragmentation (#5220), initial mux-rate measurement (#5287), and
+decoder-time admission (#5306) have landed; PSI runs and catalog linger
+remain with their current quests.
 
 ## Required
 

rs/justfile

@@ -363,6 +363,11 @@ bench-stats *args:
 bench-latency:
     bash sh/rs/nextest.sh cargo nextest run --locked --release -p moq-bench --run-ignored only -E 'test(benchmark_observation_contention)' --success-output immediate
 
+# Exact TS receiver scheduling across PID and queued-unit fanout.
+[positional-arguments]
+bench-ts-schedule *args:
+    cargo bench --locked -p moq-mux --bench ts_schedule -- "$@"
+
 # Sweep container export tracks and queued frames independently.
 [positional-arguments]
 bench-export *args:

rs/moq-mux/Cargo.toml

@@ -51,6 +51,10 @@ tracing-test = { workspace = true, features = ["no-env-filter"] }
 name = "catalog"
 harness = false
 
+[[bench]]
+name = "ts_schedule"
+harness = false
+
 [[bench]]
 name = "export"
 harness = false

rs/moq-mux/benches/ts_schedule.rs

@@ -0,0 +1,69 @@
+//! TS admission and layout across PID count and access units per PID.
+
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use std::hint::black_box;
+use std::time::Duration;
+
+mod export {
+	pub const PCR_INTERVAL: Duration = Duration::from_millis(25);
+	use std::time::Duration;
+}
+#[path = "../src/container/ts/schedule.rs"]
+#[allow(dead_code)]
+mod schedule;
+
+fn packet(pid: u16) -> [u8; 188] {
+	let mut packet = [0xff; 188];
+	packet[..4].copy_from_slice(&[0x47, (pid >> 8) as u8, pid as u8, 0x10]);
+	packet
+}
+
+fn admission(c: &mut Criterion) {
+	let mut group = c.benchmark_group("ts_schedule");
+	group.measurement_time(Duration::from_secs(1));
+	group.warm_up_time(Duration::from_millis(200));
+	group.sample_size(10);
+	for pids in [1u16, 4, 16, 64] {
+		for frames in [1u128, 16, 64] {
+			group.bench_function(BenchmarkId::new(pids.to_string(), frames), |b| {
+				b.iter(|| {
+					let mut schedule = schedule::Schedule::new(Duration::from_millis(200));
+					schedule.set_rate(Some(100_000_000));
+					schedule.set_clock(1);
+					for pid in 1..=pids {
+						schedule.set_buffer(
+							pid,
+							schedule::Buffer {
+								rate: 2_000_000,
+								size: Some(1_000_000),
+								transport: 512,
+							},
+						);
+					}
+					for frame in 0..frames {
+						for pid in 1..=pids {
+							let at = (999 + frame * 40) * 1_000_000;
+							schedule
+								.push(
+									pid,
+									schedule::Timing {
+										by: at,
+										decode: at + 1_000_000,
+									},
+									packet(pid).repeat(10),
+									false,
+								)
+								.unwrap();
+						}
+					}
+					while let Some(slot) = schedule.next(None).unwrap() {
+						black_box(slot.layout(&packet(1), 100, &packet(0x1fff)));
+					}
+				});
+			});
+		}
+	}
+	group.finish();
+}
+criterion_group!(benches, admission);
+criterion_main!(benches);

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

@@ -480,6 +480,7 @@ impl Timing {
 		Some(Buffer {
 			rate: rate.min(level.leak),
 			size: Some(usize::try_from(cpb / 8).unwrap_or(usize::MAX)),
+			transport: 512,
 		})
 	}
 
@@ -1803,8 +1804,7 @@ impl<E: catalog::Catalog> Export<E> {
 			es_info,
 		};
 
-		self.schedule.set_buffer(0, Buffer::SYSTEM);
-		self.schedule.set_buffer(pmt_pid, Buffer::SYSTEM);
+		self.schedule.set_tables(pmt_pid);
 		self.schedule.set_clock(pcr_pid);
 		for track in self.tracks.values() {
 			self.liveness.register(track.pid);
@@ -1978,7 +1978,15 @@ impl<E: catalog::Catalog> Export<E> {
 			self.schedule.set_buffer(pid, buffer);
 		}
 		let by = decode.as_nanos().saturating_sub(track.drain().as_nanos());
-		self.schedule.push(pid, by, out, keyframe);
+		self.schedule.push(
+			pid,
+			super::schedule::Timing {
+				by,
+				decode: decode.as_nanos(),
+			},
+			out,
+			keyframe,
+		)?;
 		Ok(())
 	}
 
@@ -2499,7 +2507,11 @@ fn audio_buffer(config: &AudioConfig, kind: &Kind) -> Buffer {
 		Kind::Eac3 => (2_000_000, 12_896),
 		_ => (2_000_000, 3_584),
 	};
-	Buffer { rate, size: Some(size) }
+	Buffer {
+		rate,
+		size: Some(size),
+		transport: 512,
+	}
 }
 
 /// The DVB AC-3 descriptor's tag (ETSI EN 300 468 6.2.1), marking AC-3 as private data.
@@ -2509,8 +2521,8 @@ const AC3_DESCRIPTOR: u8 = 0x6a;
 const TELETEXT_DESCRIPTOR: u8 = 0x56;
 
 /// A verbatim stream's buffers, for the kinds its descriptors identify: AC-3 as DVB private
-/// data (ATSC A/52 Annex A 5.4), and teletext, whose transport buffer drains at 6.75 Mb/s
-/// (ETSI EN 300 472 5). Any other stream has none the schedule knows.
+/// data (ATSC A/52 Annex A 5.4), and teletext, whose 480-byte transport buffer drains at
+/// 6.75 Mb/s (ETSI EN 300 472 5). Any other stream has none the schedule knows.
 fn verbatim_buffer(kind: &Kind, descriptors: &[catalog::Descriptor]) -> Option<Buffer> {
 	let Kind::Verbatim {
 		stream_type: 0x06,
@@ -2524,10 +2536,12 @@ fn verbatim_buffer(kind: &Kind, descriptors: &[catalog::Descriptor]) -> Option<B
 		AC3_DESCRIPTOR => Some(Buffer {
 			rate: 2_000_000,
 			size: Some(5_696),
+			transport: 512,
 		}),
 		TELETEXT_DESCRIPTOR => Some(Buffer {
 			rate: 6_750_000,
 			size: None,
+			transport: 480,
 		}),
 		_ => None,
 	})
@@ -2999,6 +3013,7 @@ mod tests {
 			Some(Buffer {
 				rate: 6_750_000,
 				size: None,
+				transport: 480,
 			})
 		);
 		assert_eq!(verbatim_buffer(&kind, &[descriptor(0x0a)]), None);