From 0ef8efbf9a4382b2d2e0168ffb9a1fed5512048c Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Wed, 28 Oct 2026 00:01:51 -0700
Subject: [PATCH 1/8] chore(quest): claim
 quest/m1/ts-reliability/export-si-order.md

Quest-Claim: d86ddedb2a9119ed

From 4e158769faba01f84bdbb30dba03a1ce07ee57e5 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Wed, 28 Oct 2026 02:15:51 -0700
Subject: [PATCH 2/8] fix(moq-mux): open a TS export with its PMT and a PCR
 interval

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-mux/src/container/ts/export.rs      |   5 +
 rs/moq-mux/src/container/ts/export_test.rs | 107 ++++++++++++++++++++-
 rs/moq-mux/src/container/ts/schedule.rs    |  65 +++++++++++--
 3 files changed, 168 insertions(+), 9 deletions(-)

diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs
index 7d7360e73d7..f001e199c37 100644
--- a/rs/moq-mux/src/container/ts/export.rs
+++ b/rs/moq-mux/src/container/ts/export.rs
@@ -2757,6 +2757,11 @@ where
 
 		self.program.schedule.set_tables(pmt_pid);
 		self.program.schedule.set_clock(pcr_pid);
+		// A reader of the first tables has no program clock yet. Tables that take over from
+		// an earlier program's flag a break in the clock it already reads instead.
+		if !self.program.pcr_discontinuity {
+			self.program.schedule.open();
+		}
 		for track in self.program.tracks.values() {
 			self.program.liveness.register(track.pid);
 		}
diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs
index 63a4e9c03e6..37a42c20b83 100644
--- a/rs/moq-mux/src/container/ts/export_test.rs
+++ b/rs/moq-mux/src/container/ts/export_test.rs
@@ -3474,6 +3474,109 @@ async fn service_layer_survives_roundtrip() {
 	assert_eq!(program2.pmt_pid, program.pmt_pid, "PMT PID survived");
 }
 
+/// An export leads with its PAT and PMT, then a PCR interval on the clock they name, then
+/// its SI and media: a re-import places every SI table's first emission instead of dropping
+/// it until it repeats, padded or not, and every PES with it.
+#[tokio::test(start_paused = true)]
+async fn a_reimport_keeps_the_first_si() {
+	let data = include_bytes!("test_data/bbb.ts");
+	let nit = make_long_section(0x40, 0x1234, 1, 0, 0, &[0xf0, 0x00]);
+	// Past bbb's PCRs, where its import keeps SI: bbb's own SDT precedes them.
+	let mut input = data.to_vec();
+	input.extend_from_slice(&si_packet(0x0010, &nit));
+	let mut sdt = data[..188].to_vec();
+	sdt[3] = (sdt[3] & 0xf0) | ((sdt[3] + 1) & 0x0f);
+	input.extend_from_slice(&sdt);
+	let sdt = ts_sections(&sdt, 0x0011);
+
+	// Each export reads an import of its own.
+	for rate in [None, Some(5_000_000)] {
+		let mut broadcast = moq_net::broadcast::Info::new().produce();
+		let consumer = broadcast.consume();
+		let catalog = crate::catalog::Producer::new(
+			&mut broadcast,
+			crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
+		)
+		.unwrap();
+		let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
+		import.decode(&BytesMut::from(&input[..])).unwrap();
+		import.finish().unwrap();
+		let units = |stats: super::stats::Snapshot| {
+			stats
+				.streams
+				.iter()
+				.filter(|(_, stream)| stream.units > 0)
+				.map(|(pid, stream)| (*pid, stream.units))
+				.collect::<Vec<_>>()
+		};
+		let imported = units(import.stats());
+
+		let mut export = open_ts(
+			crate::source::announced(&consumer).await,
+			crate::catalog::CatalogFormat::Hang,
+		)
+		.await
+		.unwrap();
+		if let Some(rate) = rate {
+			export = export.with_mux_rate(rate);
+		}
+		let ts = drain_with(export).await;
+		assert_packet_aligned(&ts);
+
+		// Nothing but the tables and the clock goes out before the PMT and two PCRs after it.
+		let packets = ts.as_chunks::<188>().0;
+		let pid = |packet: &[u8; 188]| u16::from(packet[1] & 0x1f) << 8 | u16::from(packet[2]);
+		let pmt = packets.iter().position(|packet| pid(packet) == 0x1000).expect("a PMT");
+		let clock = packets
+			.iter()
+			.enumerate()
+			.skip(pmt)
+			.filter(|(_, packet)| super::import::pcr(packet).is_some())
+			.nth(1)
+			.expect("two PCRs after the PMT")
+			.0;
+		let first = packets
+			.iter()
+			.position(|packet| ![0, 0x1000, 0x1fff].contains(&pid(packet)) && packet[3] & 0x10 != 0)
+			.expect("SI and media");
+		assert!(
+			first >= clock,
+			"{rate:?}: a payload at packet {first}, the PMT at {pmt} and its second PCR at {clock}"
+		);
+
+		let mut broadcast2 = moq_net::broadcast::Info::new().produce();
+		let consumer2 = broadcast2.consume();
+		let catalog2 = crate::catalog::Producer::new(
+			&mut broadcast2,
+			crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
+		)
+		.unwrap();
+		let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
+		import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
+		import2.finish().unwrap();
+		let stats = import2.stats();
+		for (pid, stream) in &stats.streams {
+			assert_eq!(
+				stream.unplaced, 0,
+				"{rate:?}: PID {pid:#x} dropped what it could not place"
+			);
+		}
+		assert_eq!(units(stats), imported, "{rate:?}: every PES");
+
+		let si = catalog2.snapshot().ext.mpegts.si;
+		assert_eq!(
+			read_si_sections(&consumer2, &si[&0x0010][&0x40].track).await,
+			vec![Bytes::from(nit.clone())],
+			"{rate:?}: the NIT"
+		);
+		assert_eq!(
+			read_si_sections(&consumer2, &si[&0x0011][&0x42].track).await,
+			sdt,
+			"{rate:?}: the SDT"
+		);
+	}
+}
+
 /// Each SI PID must be re-emitted on its own interval, independently of the PSI cadence
 /// and of video keyframes. The fixtures are a fraction of a second long, so nothing else
 /// distinguishes a correct interval from "emitted once and never again": this builds a
@@ -4739,9 +4842,7 @@ async fn a_selected_programs_si_survives_export() {
 	);
 }
 
-/// The sections `ts` carries on `pid`, each whole in one packet. Read off the wire rather
-/// than by a re-import: SI an export leads with precedes its PCRs, so an import drops it
-/// until it repeats.
+/// The sections `ts` carries on `pid`, each whole in one packet, read off the wire.
 fn ts_sections(ts: &[u8], pid: u16) -> Vec<Bytes> {
 	ts.chunks(188)
 		.filter(|p| (u16::from(p[1] & 0x1f) << 8 | u16::from(p[2])) == pid && p[1] & 0x40 != 0)
diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs
index 1976088c71e..6ce58f204cc 100644
--- a/rs/moq-mux/src/container/ts/schedule.rs
+++ b/rs/moq-mux/src/container/ts/schedule.rs
@@ -30,6 +30,11 @@ const PAYLOAD: usize = TsPacket::SIZE - 4;
 const BACKFILL: u128 = 40;
 /// The longest a transport buffer may go without emptying (ISO 13818-1 2.4.2.3).
 const TB_EMPTY: Duration = Duration::from_secs(1);
+/// How many slots a timeline that [opens](Schedule::open) lays its first program tables ahead
+/// of everything else. A reader learns the clock PID from the PMT, so it times nothing by the
+/// PCR opening their slot; the next slot's PCR is the first it reads, and the one after closes
+/// the first interval, which it needs to place SI and whole PES on the program clock.
+const OPENING: u128 = 2;
 /// One tick of the system clock, in whole nanoseconds rounded up.
 const TICK: i128 = 1_000_000_000_u128.div_ceil(SYSTEM_CLOCK) as i128;
 /// The system clock in Hz, the unit of a PCR.
@@ -589,6 +594,12 @@ pub(super) struct Schedule {
 	seq: u64,
 	/// Tables to go out ahead of the next unit pushed.
 	lead: Option<Tables>,
+	/// The timeline opens with its clock ([`Self::open`]) and its first tables are still to
+	/// go out: they go [`OPENING`] slots ahead of their own release.
+	opening: bool,
+	/// The first slot a unit other than program tables may go out in: [`OPENING`] slots after
+	/// the slot the opening tables finish in.
+	clocked: u128,
 }
 
 impl Schedule {
@@ -610,9 +621,20 @@ impl Schedule {
 			ended: false,
 			seq: 0,
 			lead: None,
+			opening: false,
+			clocked: 0,
 		}
 	}
 
+	/// Open the timeline with its clock: its first program tables go out ahead of everything
+	/// else, and nothing else goes out until a PCR interval has followed them ([`OPENING`]),
+	/// so a reader of the stream from its first byte places all of it. A unit due before
+	/// then goes out as soon as it can, as one pushed late does. For a timeline no reader has
+	/// a clock for yet; one that continues an earlier output's clock needs no opening.
+	pub fn open(&mut self) {
+		self.opening = true;
+	}
+
 	/// Whether every source has ended, so what is queued still goes out, late if it must.
 	/// A source that resumes (a same-instance return) holds its deadlines again.
 	pub fn set_ended(&mut self, ended: bool) {
@@ -868,7 +890,7 @@ impl Schedule {
 			packets,
 			sent: 0,
 			keyframe: false,
-			late: self.next.is_some_and(|next| due < next),
+			late: self.next.is_some_and(|next| due < next) || (tables.is_none() && due < self.clocked),
 			place: None,
 			held: 0,
 			frames: Vec::new(),
@@ -894,16 +916,29 @@ impl Schedule {
 		held
 	}
 
-	/// Put in force what the tables of unit `i` carry, as their last packet goes out, if it is one.
-	fn enforce(&mut self, i: usize) -> bool {
+	/// Put in force what the tables of unit `i` carry, as their last packet goes out in slot
+	/// `index`, if it is one.
+	fn enforce(&mut self, i: usize, index: u128) -> bool {
 		let Some(program) = &self.units[i].tables else {
 			return false;
 		};
 		self.clock = Some(program.clock);
 		self.rate = program.rate;
+		if std::mem::take(&mut self.opening) {
+			self.hold_until(index + OPENING);
+		}
 		true
 	}
 
+	/// Hold every unit other than program tables until slot `clocked`: one due before it goes
+	/// out as soon as it can.
+	fn hold_until(&mut self, clocked: u128) {
+		self.clocked = self.clocked.max(clocked);
+		for unit in self.units.iter_mut().filter(|unit| unit.tables.is_none()) {
+			unit.late |= unit.due < self.clocked;
+		}
+	}
+
 	/// Whether a unit is still to go out.
 	#[cfg(test)]
 	pub fn queued(&self) -> bool {
@@ -930,6 +965,7 @@ impl Schedule {
 		self.last = None;
 		self.ended = false;
 		self.lead = None;
+		self.clocked = 0;
 	}
 
 	/// The first slot `unit` may go out in: [`ahead`] of its due slot, and never before the
@@ -939,7 +975,20 @@ impl Schedule {
 	/// multiplex rate the one whose span holds it, where [`Self::padded`] times each packet,
 	/// and otherwise the first that opens at or after it, as nothing times the packets
 	/// within an unpadded slot. Slot `index` opens one slot behind its boundary ([`grid_pcr`]).
+	///
+	/// On a timeline that [opens](Self::open), its first tables go out [`OPENING`] slots sooner,
+	/// and nothing else until [`OPENING`] slots after the slot they finish in.
 	fn release(&self, unit: &Unit) -> u128 {
+		let release = self.own_release(unit);
+		match unit.tables {
+			Some(_) if self.opening => release.saturating_sub(OPENING),
+			Some(_) => release,
+			None => release.max(self.clocked),
+		}
+	}
+
+	/// [`Self::release`] without the opening.
+	fn own_release(&self, unit: &Unit) -> u128 {
 		if let Some((place, _)) = unit.place {
 			let slot = PCR_INTERVAL.as_nanos();
 			return match self.rate {
@@ -991,6 +1040,9 @@ impl Schedule {
 		if known.is_some_and(|known| index >= known) {
 			return Ok(None);
 		}
+		if self.opening && self.next.is_none() {
+			self.hold_until(index + OPENING);
+		}
 		self.next = Some(index + 1);
 		if let Some(rate) = self.rate {
 			return self.padded(index, rate).map(Some);
@@ -1044,7 +1096,7 @@ impl Schedule {
 			unit.sent += take;
 		}
 		let tables = (0..self.units.len())
-			.filter(|&i| self.units[i].remaining() == 0 && self.enforce(i))
+			.filter(|&i| self.units[i].remaining() == 0 && self.enforce(i, index))
 			.count();
 		for unit in self.units.iter().filter(|unit| unit.remaining() == 0) {
 			self.decoding.extend(unit.leaving());
@@ -1172,7 +1224,8 @@ impl Schedule {
 			while let Some(Reverse((by, lane))) = ready.pop() {
 				let i = *lanes[lane].front().unwrap();
 				let unit = &self.units[i];
-				if holding.get(&unit.pid).is_some_and(|&seq| seq < unit.seq) {
+				// The opening tables finishing in this slot hold back what was released beside them.
+				if holding.get(&unit.pid).is_some_and(|&seq| seq < unit.seq) || self.release(unit) > index {
 					passed.push(Reverse((by, lane)));
 					continue;
 				}
@@ -1357,7 +1410,7 @@ impl Schedule {
 				}
 				if unit.remaining() == 0 {
 					lanes[lane].pop_front();
-					if self.enforce(i) {
+					if self.enforce(i, index) {
 						out.tables += 1;
 						holding = self.holding(|i| self.units[i].remaining());
 					}

From 0b77432ca537ebedd5890a05c2aa53d774c0a545 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Wed, 28 Oct 2026 02:34:16 -0700
Subject: [PATCH 3/8] fix(moq-mux): hold placed units and SI for the opening's
 PCR interval

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-mux/src/container/ts/export.rs      |  12 +++
 rs/moq-mux/src/container/ts/export_test.rs |  16 +--
 rs/moq-mux/src/container/ts/schedule.rs    | 109 +++++++++++++--------
 3 files changed, 89 insertions(+), 48 deletions(-)

diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs
index f001e199c37..e112c3c5d0d 100644
--- a/rs/moq-mux/src/container/ts/export.rs
+++ b/rs/moq-mux/src/container/ts/export.rs
@@ -3136,12 +3136,24 @@ where
 				(*pid, si.active.sections().cloned().collect())
 			})
 			.collect();
+		let si = pending.first().map(|(pid, _)| *pid);
 		for (pid, sections) in pending {
 			self.program.schedule.set_buffer(pid, Buffer::SYSTEM);
 			for section in &sections {
 				self.write_section(&mut out, pid, section)?;
 			}
 		}
+		// A reader drops SI it cannot place on the program clock, so while the opening's PCR
+		// interval may still be ahead of the unit this frame goes in, the SI goes on its own,
+		// placed at the frame's decode time, rather than hold back the media it would lead. A
+		// placed frame waits for the clock anyway.
+		if let Some(pid) = si.filter(|_| !kind.placed() && self.program.schedule.clocking()) {
+			let place = decode.as_nanos();
+			let si = std::mem::take(&mut out);
+			self.program
+				.schedule
+				.push_placed(pid, (place, place), 1, si, false, Vec::new())?;
+		}
 
 		// The tables lead the unit that carries this frame, behind whatever went before it.
 		if kind.placed() {
diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs
index 37a42c20b83..829ccc2ae73 100644
--- a/rs/moq-mux/src/container/ts/export_test.rs
+++ b/rs/moq-mux/src/container/ts/export_test.rs
@@ -3474,9 +3474,9 @@ async fn service_layer_survives_roundtrip() {
 	assert_eq!(program2.pmt_pid, program.pmt_pid, "PMT PID survived");
 }
 
-/// An export leads with its PAT and PMT, then a PCR interval on the clock they name, then
-/// its SI and media: a re-import places every SI table's first emission instead of dropping
-/// it until it repeats, padded or not, and every PES with it.
+/// An export leads with its PAT and PMT, then a PCR interval on the clock they name, before
+/// its SI: a re-import places every SI table's first emission instead of dropping it until it
+/// repeats, padded or not, and every PES with it.
 #[tokio::test(start_paused = true)]
 async fn a_reimport_keeps_the_first_si() {
 	let data = include_bytes!("test_data/bbb.ts");
@@ -3523,7 +3523,7 @@ async fn a_reimport_keeps_the_first_si() {
 		let ts = drain_with(export).await;
 		assert_packet_aligned(&ts);
 
-		// Nothing but the tables and the clock goes out before the PMT and two PCRs after it.
+		// No SI goes out before the PMT and two PCRs after it.
 		let packets = ts.as_chunks::<188>().0;
 		let pid = |packet: &[u8; 188]| u16::from(packet[1] & 0x1f) << 8 | u16::from(packet[2]);
 		let pmt = packets.iter().position(|packet| pid(packet) == 0x1000).expect("a PMT");
@@ -3537,11 +3537,11 @@ async fn a_reimport_keeps_the_first_si() {
 			.0;
 		let first = packets
 			.iter()
-			.position(|packet| ![0, 0x1000, 0x1fff].contains(&pid(packet)) && packet[3] & 0x10 != 0)
-			.expect("SI and media");
+			.position(|packet| [0x0010, 0x0011].contains(&pid(packet)))
+			.expect("SI");
 		assert!(
-			first >= clock,
-			"{rate:?}: a payload at packet {first}, the PMT at {pmt} and its second PCR at {clock}"
+			first > clock,
+			"{rate:?}: SI at packet {first}, the PMT at {pmt} and its second PCR at {clock}"
 		);
 
 		let mut broadcast2 = moq_net::broadcast::Info::new().produce();
diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs
index 6ce58f204cc..e9810db951f 100644
--- a/rs/moq-mux/src/container/ts/schedule.rs
+++ b/rs/moq-mux/src/container/ts/schedule.rs
@@ -30,10 +30,11 @@ const PAYLOAD: usize = TsPacket::SIZE - 4;
 const BACKFILL: u128 = 40;
 /// The longest a transport buffer may go without emptying (ISO 13818-1 2.4.2.3).
 const TB_EMPTY: Duration = Duration::from_secs(1);
-/// How many slots a timeline that [opens](Schedule::open) lays its first program tables ahead
-/// of everything else. A reader learns the clock PID from the PMT, so it times nothing by the
-/// PCR opening their slot; the next slot's PCR is the first it reads, and the one after closes
-/// the first interval, which it needs to place SI and whole PES on the program clock.
+/// How many slots after the one the first program tables finish in a timeline that
+/// [opens](Schedule::open) places nothing. A reader learns the clock PID from the PMT, so it
+/// times nothing by the PCR opening their slot; the next slot's PCR is the first it reads, and
+/// the one after closes the first interval, which it needs to place a section or whole PES on
+/// the program clock.
 const OPENING: u128 = 2;
 /// One tick of the system clock, in whole nanoseconds rounded up.
 const TICK: i128 = 1_000_000_000_u128.div_ceil(SYSTEM_CLOCK) as i128;
@@ -594,11 +595,10 @@ pub(super) struct Schedule {
 	seq: u64,
 	/// Tables to go out ahead of the next unit pushed.
 	lead: Option<Tables>,
-	/// The timeline opens with its clock ([`Self::open`]) and its first tables are still to
-	/// go out: they go [`OPENING`] slots ahead of their own release.
+	/// The timeline opens ([`Self::open`]) and its first tables are still to go out.
 	opening: bool,
-	/// The first slot a unit other than program tables may go out in: [`OPENING`] slots after
-	/// the slot the opening tables finish in.
+	/// The first slot a placed unit may go out in: [`OPENING`] slots after the one the first
+	/// tables finish in, on a timeline that opens.
 	clocked: u128,
 }
 
@@ -626,15 +626,18 @@ impl Schedule {
 		}
 	}
 
-	/// Open the timeline with its clock: its first program tables go out ahead of everything
-	/// else, and nothing else goes out until a PCR interval has followed them ([`OPENING`]),
-	/// so a reader of the stream from its first byte places all of it. A unit due before
-	/// then goes out as soon as it can, as one pushed late does. For a timeline no reader has
-	/// a clock for yet; one that continues an earlier output's clock needs no opening.
+	/// Open the timeline for a reader that has no program clock yet: no placed unit goes out
+	/// until a PCR interval has followed the first tables ([`OPENING`]), so the reader can
+	/// place every one. A timeline that continues a clock the reader already has needs none.
 	pub fn open(&mut self) {
 		self.opening = true;
 	}
 
+	/// Whether a unit pushed now could go out before the opening's PCR interval is out.
+	pub fn clocking(&self) -> bool {
+		self.opening || self.next.is_some_and(|next| next < self.clocked)
+	}
+
 	/// Whether every source has ended, so what is queued still goes out, late if it must.
 	/// A source that resumes (a same-instance return) holds its deadlines again.
 	pub fn set_ended(&mut self, ended: bool) {
@@ -890,7 +893,7 @@ impl Schedule {
 			packets,
 			sent: 0,
 			keyframe: false,
-			late: self.next.is_some_and(|next| due < next) || (tables.is_none() && due < self.clocked),
+			late: self.next.is_some_and(|next| due < next),
 			place: None,
 			held: 0,
 			frames: Vec::new(),
@@ -925,20 +928,11 @@ impl Schedule {
 		self.clock = Some(program.clock);
 		self.rate = program.rate;
 		if std::mem::take(&mut self.opening) {
-			self.hold_until(index + OPENING);
+			self.clocked = self.clocked.max(index + OPENING);
 		}
 		true
 	}
 
-	/// Hold every unit other than program tables until slot `clocked`: one due before it goes
-	/// out as soon as it can.
-	fn hold_until(&mut self, clocked: u128) {
-		self.clocked = self.clocked.max(clocked);
-		for unit in self.units.iter_mut().filter(|unit| unit.tables.is_none()) {
-			unit.late |= unit.due < self.clocked;
-		}
-	}
-
 	/// Whether a unit is still to go out.
 	#[cfg(test)]
 	pub fn queued(&self) -> bool {
@@ -975,26 +969,15 @@ impl Schedule {
 	/// multiplex rate the one whose span holds it, where [`Self::padded`] times each packet,
 	/// and otherwise the first that opens at or after it, as nothing times the packets
 	/// within an unpadded slot. Slot `index` opens one slot behind its boundary ([`grid_pcr`]).
-	///
-	/// On a timeline that [opens](Self::open), its first tables go out [`OPENING`] slots sooner,
-	/// and nothing else until [`OPENING`] slots after the slot they finish in.
+	/// On a timeline that [opens](Self::open), it waits for the opening's PCR interval too.
 	fn release(&self, unit: &Unit) -> u128 {
-		let release = self.own_release(unit);
-		match unit.tables {
-			Some(_) if self.opening => release.saturating_sub(OPENING),
-			Some(_) => release,
-			None => release.max(self.clocked),
-		}
-	}
-
-	/// [`Self::release`] without the opening.
-	fn own_release(&self, unit: &Unit) -> u128 {
 		if let Some((place, _)) = unit.place {
 			let slot = PCR_INTERVAL.as_nanos();
-			return match self.rate {
+			let release = match self.rate {
 				Some(_) => place / slot + 1,
 				None => place.div_ceil(slot) + 1,
 			};
+			return release.max(self.clocked);
 		}
 		if unit.late {
 			return 0;
@@ -1040,8 +1023,10 @@ impl Schedule {
 		if known.is_some_and(|known| index >= known) {
 			return Ok(None);
 		}
+		// The first tables finish in the first slot unless their buffers hold them back, and a
+		// placed unit beside them there would go out ahead of its clock.
 		if self.opening && self.next.is_none() {
-			self.hold_until(index + OPENING);
+			self.clocked = index + OPENING;
 		}
 		self.next = Some(index + 1);
 		if let Some(rate) = self.rate {
@@ -1224,7 +1209,7 @@ impl Schedule {
 			while let Some(Reverse((by, lane))) = ready.pop() {
 				let i = *lanes[lane].front().unwrap();
 				let unit = &self.units[i];
-				// The opening tables finishing in this slot hold back what was released beside them.
+				// First tables that finish past the first slot hold back what was released beside them.
 				if holding.get(&unit.pid).is_some_and(|&seq| seq < unit.seq) || self.release(unit) > index {
 					passed.push(Reverse((by, lane)));
 					continue;
@@ -1734,6 +1719,50 @@ mod tests {
 		assert_eq!(tables, 2);
 	}
 
+	/// A timeline that opens holds a placed unit until a PCR interval has followed its first
+	/// tables, while the unit they lead goes out where it would have without the opening, padded
+	/// or not.
+	#[test]
+	fn an_opening_holds_placed_units_for_a_pcr_interval() {
+		for rate in [None, Some(RATE)] {
+			let first = |open: bool| {
+				let mut schedule = Schedule::new(Duration::ZERO);
+				schedule.set_rate(rate);
+				schedule.set_tables(100);
+				schedule.set_clock(1);
+				schedule.set_buffer(1, VIDEO);
+				if open {
+					schedule.open();
+				}
+				assert_eq!(schedule.clocking(), open);
+				let mut section = unit(7, 1);
+				section[1] |= 0x40;
+				schedule.push_placed(7, (0, 0), 1, section, false, Vec::new()).unwrap();
+				schedule.lead(Tables {
+					packets: [unit(0, 1), unit(100, 1)].concat(),
+					clock: 1,
+					rate,
+					added: vec![1],
+				});
+				schedule.push(1, timing(10), unit(1, 3), true).unwrap();
+				let slots = drain(&mut schedule);
+				assert!(!schedule.clocking());
+				let on = |pid: u16| {
+					slots
+						.iter()
+						.find(|(_, pids, _)| pids.contains_key(&pid))
+						.map(|&(index, ..)| index)
+				};
+				(on(100).unwrap(), on(1).unwrap(), on(7).unwrap())
+			};
+			let (tables, media, placed) = first(true);
+			assert_eq!(placed, tables + OPENING, "{rate:?}");
+			let (_, before, early) = first(false);
+			assert_eq!(media, before, "{rate:?}: the media keeps its slot");
+			assert!(early < placed, "{rate:?}: without the opening it goes in slot {early}");
+		}
+	}
+
 	/// The pointer byte goes just before placement, so it cannot cost a section another packet.
 	#[test]
 	fn placed_first_byte_accounts_for_the_section_pointer() {

From 69cfd46247e8466b50cd5d1c76efe86645d39b8c Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Wed, 28 Oct 2026 03:02:39 -0700
Subject: [PATCH 4/8] fix(moq-mux): hold a TS export's first SI for its clock
 instead of its media

The opening holds only what a reader places on the program clock: placed
PES and sections wait in the schedule, and the SI muxed before the PCR
interval is out goes as packets of its own once it is. Decoded media keeps
its slots, deadlines, and pacing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 quest/m1/ts-reliability/README.md          |  1 -
 quest/m1/ts-reliability/export-si-order.md | 27 --------
 rs/moq-mux/src/container/ts/export.rs      | 18 +++---
 rs/moq-mux/src/container/ts/export_test.rs | 34 +++++-----
 rs/moq-mux/src/container/ts/schedule.rs    | 73 +++++++++++++++++++---
 5 files changed, 89 insertions(+), 64 deletions(-)
 delete mode 100644 quest/m1/ts-reliability/export-si-order.md

diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md
index c908d9e6d33..ca28fd3c8fb 100644
--- a/quest/m1/ts-reliability/README.md
+++ b/quest/m1/ts-reliability/README.md
@@ -100,7 +100,6 @@ catalog linger (#5305).
 
 - [PAT changes](/quest/m1/ts-reliability/pat-lifecycle.md) - a revised PAT never leaves a broadcast meaning another program
 - [TS workflow passes](/quest/m1/ts-reliability/validation-green.md) - the scheduled TS lanes pass on main again, with no lane loosened
-- [Export SI order](/quest/m1/ts-reliability/export-si-order.md) - a TS export sends SI and media after its PMT and a PCR interval, so a hop keeps first emissions
 - [A stitched switch returns to --delay](/quest/m1/ts-reliability/stitch-catch-up.md) - `--stitch` drops the old program's tail once the replacement acquires, so latency settles back to `--delay`
 
 ## Related
diff --git a/quest/m1/ts-reliability/export-si-order.md b/quest/m1/ts-reliability/export-si-order.md
deleted file mode 100644
index a5a2026e83f..00000000000
--- a/quest/m1/ts-reliability/export-si-order.md
+++ /dev/null
@@ -1,27 +0,0 @@
-# [S] TS export leads with its PMT and PCR
-
-## Goal
-
-A TS export holds its first SI tables and media until the PMT and one PCR
-interval are out, so an export-to-import hop keeps each table's first emission instead of
-dropping it until the table repeats.
-
-## Plan
-
-Since [moq-dev/moq#6757](https://github.com/moq-dev/moq/pull/6757), TS import
-admits SI only once a valid program-clock interval exists. The export writes
-its first SI ahead of its PMT and PCRs, so the importer drops it. Order the
-export's first output as PAT, PMT, a PCR interval, then SI. A PCR interval
-here is what `si.rs` needs to place a section: one forward PCR step of at most
-`MAX_INTERVAL`, not the importer's two-second catalog window. Test it with an
-export-to-import round trip that sees every SI table's first emission.
-
-The first media PES also waits for that PCR, so a reader of our own export
-never sees media it cannot place. The round trip also checks that the first
-PES follows the PCR and is imported.
-
-Not planned (2026-10-25, maintainer): bounding SI aging against a source that
-claims a huge mux rate through valid PCR steps. That needs a program-clock
-floor per input byte, a new contract.
-
-Public API: none. Wire: none.
diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs
index e112c3c5d0d..7f3a069edbd 100644
--- a/rs/moq-mux/src/container/ts/export.rs
+++ b/rs/moq-mux/src/container/ts/export.rs
@@ -103,7 +103,9 @@ fn sanitize_mux_rate(rate: u64) -> Option<u64> {
 /// Use [`next`](Self::next) to pull one [`Frame`] per PCR grid slot: its `payload`
 /// is the TS packets belonging to that slot, stamped at the slot's boundary. The
 /// leading PAT/PMT rides on the first frame (so it inherits a real timestamp), and
-/// is re-emitted at video keyframes and periodically for mid-stream tune-in.
+/// is re-emitted at video keyframes and periodically for mid-stream tune-in. SI and
+/// whole PES or sections wait until a PCR interval has followed it, since a reader
+/// places them on the program clock and the PMT is what names it.
 /// Returns `None` when the broadcast ends.
 ///
 /// A rendition the catalog moves to another track (a new `track` or `broadcast`) goes on out
@@ -3136,7 +3138,6 @@ where
 				(*pid, si.active.sections().cloned().collect())
 			})
 			.collect();
-		let si = pending.first().map(|(pid, _)| *pid);
 		for (pid, sections) in pending {
 			self.program.schedule.set_buffer(pid, Buffer::SYSTEM);
 			for section in &sections {
@@ -3144,15 +3145,10 @@ where
 			}
 		}
 		// A reader drops SI it cannot place on the program clock, so while the opening's PCR
-		// interval may still be ahead of the unit this frame goes in, the SI goes on its own,
-		// placed at the frame's decode time, rather than hold back the media it would lead. A
-		// placed frame waits for the clock anyway.
-		if let Some(pid) = si.filter(|_| !kind.placed() && self.program.schedule.clocking()) {
-			let place = decode.as_nanos();
-			let si = std::mem::take(&mut out);
-			self.program
-				.schedule
-				.push_placed(pid, (place, place), 1, si, false, Vec::new())?;
+		// interval may still be ahead of the unit this frame goes in, the schedule holds the SI
+		// for it rather than the media it would lead.
+		if self.program.schedule.clocking() {
+			self.program.schedule.push_clocked(std::mem::take(&mut out));
 		}
 
 		// The tables lead the unit that carries this frame, behind whatever went before it.
diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs
index 829ccc2ae73..d02dd31685c 100644
--- a/rs/moq-mux/src/container/ts/export_test.rs
+++ b/rs/moq-mux/src/container/ts/export_test.rs
@@ -126,11 +126,6 @@ where
 	(out, exporter.stats())
 }
 
-/// An adaptation-field-only single-packet frame: the exporter's PCR carriage.
-fn is_pcr_frame(frame: &Frame) -> bool {
-	frame.payload.len() == 188 && frame.payload[3] & 0x30 == 0x20
-}
-
 fn assert_packet_aligned(ts: &[u8]) {
 	assert!(!ts.is_empty(), "no TS output");
 	assert_eq!(ts.len() % 188, 0, "output not a whole number of 188-byte packets");
@@ -5647,12 +5642,17 @@ async fn late_join_matches_a_running_exporter_without_video() {
 	let (a, b) = export_twice(false).await;
 
 	// The joiner's first frame leads with PAT/PMT, while the running exporter has no
-	// reason to repeat them there. It can spread over several PCR slices, so compare
-	// from the next PAT/PMT refresh, which both exporters anchor to the media grid.
+	// reason to repeat them there, and its SDT waits for a PCR interval after them. They can
+	// spread over several PCR slices, so compare from the next PAT/PMT refresh past that SDT,
+	// which both exporters anchor to the media grid.
+	let sdt = b
+		.iter()
+		.find(|frame| count_pid(std::slice::from_ref(*frame), 0x0011) > 0)
+		.expect("the joiner's SDT")
+		.timestamp;
 	let from = b
 		.iter()
-		.filter(|frame| count_pid(std::slice::from_ref(*frame), 0) > 0)
-		.nth(1)
+		.find(|frame| frame.timestamp > sdt && count_pid(std::slice::from_ref(*frame), 0) > 0)
 		.expect("a periodic PAT/PMT refresh")
 		.timestamp;
 	assert_only_continuity_differs(&a, &b, from);
@@ -6475,18 +6475,18 @@ async fn si_revision_after_final_media_frame_is_flushed() {
 	)
 	.await
 	.unwrap();
-	let first = loop {
+	// v1 goes out with the media frame's slices, once the export's PCR interval is out.
+	let contains = |haystack: &[u8], needle: &[u8]| haystack.windows(needle.len()).any(|w| w == needle);
+	loop {
 		let frame = tokio::time::timeout(DRAIN, exporter.next())
 			.await
-			.expect("the only media frame")
+			.expect("v1 goes out (control)")
 			.unwrap()
-			.unwrap();
-		if !is_pcr_frame(&frame) {
-			break frame;
+			.expect("v1 goes out (control)");
+		if contains(&frame.payload, &sdt_v1) {
+			break;
 		}
-	};
-	let contains = |haystack: &[u8], needle: &[u8]| haystack.windows(needle.len()).any(|w| w == needle);
-	assert!(contains(&first.payload, &sdt_v1), "v1 rode the media frame (control)");
+	}
 
 	// The revision lands after the last media frame; only the trailing flush can
 	// carry it. Closing the catalog lets the exporter reach end of stream.
diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs
index e9810db951f..1d982d9ab8a 100644
--- a/rs/moq-mux/src/container/ts/schedule.rs
+++ b/rs/moq-mux/src/container/ts/schedule.rs
@@ -30,11 +30,11 @@ const PAYLOAD: usize = TsPacket::SIZE - 4;
 const BACKFILL: u128 = 40;
 /// The longest a transport buffer may go without emptying (ISO 13818-1 2.4.2.3).
 const TB_EMPTY: Duration = Duration::from_secs(1);
-/// How many slots after the one the first program tables finish in a timeline that
-/// [opens](Schedule::open) places nothing. A reader learns the clock PID from the PMT, so it
-/// times nothing by the PCR opening their slot; the next slot's PCR is the first it reads, and
-/// the one after closes the first interval, which it needs to place a section or whole PES on
-/// the program clock.
+/// How many slots after the one its first program tables finish in a timeline that
+/// [opens](Schedule::open) holds what a reader places on the program clock. The reader learns
+/// the clock PID from the PMT, so it times nothing by the PCR opening their slot; the next
+/// slot's PCR is the first it reads, and the one after closes the first interval, which it
+/// needs to place a section or whole PES.
 const OPENING: u128 = 2;
 /// One tick of the system clock, in whole nanoseconds rounded up.
 const TICK: i128 = 1_000_000_000_u128.div_ceil(SYSTEM_CLOCK) as i128;
@@ -600,6 +600,9 @@ pub(super) struct Schedule {
 	/// The first slot a placed unit may go out in: [`OPENING`] slots after the one the first
 	/// tables finish in, on a timeline that opens.
 	clocked: u128,
+	/// Packets a reader places on the program clock, held until the opening's PCR interval is
+	/// out ([`Self::push_clocked`]).
+	unclocked: Vec<u8>,
 }
 
 impl Schedule {
@@ -623,6 +626,7 @@ impl Schedule {
 			lead: None,
 			opening: false,
 			clocked: 0,
+			unclocked: Vec::new(),
 		}
 	}
 
@@ -638,6 +642,12 @@ impl Schedule {
 		self.opening || self.next.is_some_and(|next| next < self.clocked)
 	}
 
+	/// Hold `packets`, which a reader places on the program clock and which ride no unit of their
+	/// own, until the opening's PCR interval is out: they then go out first, as soon as they can.
+	pub fn push_clocked(&mut self, packets: Vec<u8>) {
+		self.unclocked.extend(packets);
+	}
+
 	/// Whether every source has ended, so what is queued still goes out, late if it must.
 	/// A source that resumes (a same-instance return) holds its deadlines again.
 	pub fn set_ended(&mut self, ended: bool) {
@@ -933,15 +943,17 @@ impl Schedule {
 		true
 	}
 
-	/// Whether a unit is still to go out.
+	/// Whether a unit, or packets held for the opening's clock, are still to go out.
 	#[cfg(test)]
 	pub fn queued(&self) -> bool {
-		!self.units.is_empty()
+		!self.units.is_empty() || !self.unclocked.is_empty()
 	}
 
 	/// Whether nothing is left to lay out: every unit sent, and the clock past its decode time.
 	pub fn is_empty(&self) -> bool {
+		// Packets held for the clock count once a slot has gone out, so slots run on to it.
 		self.units.is_empty()
+			&& (self.unclocked.is_empty() || self.next.is_none())
 			&& self
 				.horizon
 				.is_none_or(|horizon| self.next.is_some_and(|next| next > horizon))
@@ -960,6 +972,7 @@ impl Schedule {
 		self.ended = false;
 		self.lead = None;
 		self.clocked = 0;
+		self.unclocked.clear();
 	}
 
 	/// The first slot `unit` may go out in: [`ahead`] of its due slot, and never before the
@@ -1028,6 +1041,15 @@ impl Schedule {
 		if self.opening && self.next.is_none() {
 			self.clocked = index + OPENING;
 		}
+		if !self.opening && index >= self.clocked && !self.unclocked.is_empty() {
+			let packets = std::mem::take(&mut self.unclocked);
+			let by = (index + 1) * PCR_INTERVAL.as_nanos();
+			// They follow no unit, so the next one pushed keeps its own start.
+			let last = self.last;
+			self.enqueue(pid(&packets), Timing { by, decode: by }, packets, None)
+				.late = true;
+			self.last = last;
+		}
 		self.next = Some(index + 1);
 		if let Some(rate) = self.rate {
 			return self.padded(index, rate).map(Some);
@@ -1209,7 +1231,8 @@ impl Schedule {
 			while let Some(Reverse((by, lane))) = ready.pop() {
 				let i = *lanes[lane].front().unwrap();
 				let unit = &self.units[i];
-				// First tables that finish past the first slot hold back what was released beside them.
+				// First tables that finish past the first slot hold back what was released beside
+				// them.
 				if holding.get(&unit.pid).is_some_and(|&seq| seq < unit.seq) || self.release(unit) > index {
 					passed.push(Reverse((by, lane)));
 					continue;
@@ -1763,6 +1786,40 @@ mod tests {
 		}
 	}
 
+	/// Packets held for the opening's clock go out once its PCR interval is, without holding
+	/// back the unit they were muxed ahead of, padded or not.
+	#[test]
+	fn clocked_packets_follow_the_opening_without_the_media() {
+		for rate in [None, Some(RATE)] {
+			let mut schedule = Schedule::new(Duration::ZERO);
+			schedule.set_rate(rate);
+			schedule.set_tables(100);
+			schedule.set_clock(1);
+			schedule.set_buffer(1, VIDEO);
+			schedule.set_buffer(0x11, Buffer::SYSTEM);
+			schedule.open();
+			schedule.push_clocked(unit(0x11, 1));
+			schedule.lead(Tables {
+				packets: [unit(0, 1), unit(100, 1)].concat(),
+				clock: 1,
+				rate,
+				added: vec![1],
+			});
+			schedule.push(1, timing(10), unit(1, 3), true).unwrap();
+			assert!(schedule.clocking());
+			let slots = drain(&mut schedule);
+			assert!(!schedule.clocking());
+			let on = |pid: u16| {
+				slots
+					.iter()
+					.find(|(_, pids, _)| pids.contains_key(&pid))
+					.map(|&(index, ..)| index)
+			};
+			assert_eq!((on(100), on(1)), (Some(0), Some(0)), "{rate:?}");
+			assert_eq!(on(0x11), Some(OPENING), "{rate:?}");
+		}
+	}
+
 	/// The pointer byte goes just before placement, so it cannot cost a section another packet.
 	#[test]
 	fn placed_first_byte_accounts_for_the_section_pointer() {

From a19f532f449d3ef6aa0c7bf707b5259cf2af787b Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Wed, 28 Oct 2026 08:54:48 -0700
Subject: [PATCH 5/8] fix(moq-mux): keep a TS export's opening PCR interval
 whole across a gap

When the queue emptied after the first frames, `Schedule::index` skipped to
`BACKFILL` slots before the next frame without regard for the SI held for the
opening, so a zero-delay export with a gap past a second sent that SI after a
PCR jump the reader cannot measure, and the re-import dropped it. While the
opening is pending, the skip now stops at its gate.

`clocking()` also stays true while SI is held, so a later revision muxed then
queues behind it rather than going out first.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-mux/src/container/ts/export_test.rs | 117 +++++++++++++++++++++
 rs/moq-mux/src/container/ts/schedule.rs    |  52 ++++++++-
 2 files changed, 166 insertions(+), 3 deletions(-)

diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs
index d02dd31685c..38b1f86bf31 100644
--- a/rs/moq-mux/src/container/ts/export_test.rs
+++ b/rs/moq-mux/src/container/ts/export_test.rs
@@ -3572,6 +3572,123 @@ async fn a_reimport_keeps_the_first_si() {
 	}
 }
 
+/// A gap longer than a skip after the first frame still leaves the opening's PCR interval
+/// whole: the first SI follows two PCRs a reader can measure, and a re-import places it
+/// instead of dropping it until it repeats, padded or not.
+#[tokio::test(start_paused = true)]
+async fn a_reimport_keeps_the_first_si_across_a_gap() {
+	for rate in [None, Some(5_000_000)] {
+		let mut broadcast = moq_net::broadcast::Info::new().produce();
+		let consumer = broadcast.consume();
+		let mut catalog = crate::catalog::Producer::new(
+			&mut broadcast,
+			crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
+		)
+		.unwrap();
+
+		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();
+		let mut sdt_track = broadcast.create_track("0x0011-0x42.si", timed()).unwrap();
+		sdt_track
+			.write_frame(
+				Timestamp::ZERO,
+				Bytes::from(make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 8])),
+			)
+			.unwrap();
+		{
+			let mut guard = catalog.modify().unwrap();
+			let mut cfg = VideoConfig::new(H264 {
+				profile: 0x64,
+				constraints: 0,
+				level: 0x1f,
+				inline: false,
+			});
+			cfg.container = Container::Legacy;
+			cfg.description = Some(avcc);
+			guard.video.renditions.insert(name.clone(), cfg);
+			guard.ext.mpegts.si.entry(0x0011).or_default().insert(
+				0x42,
+				tscat::SiEntry {
+					track: "0x0011-0x42.si".to_string(),
+					interval: Some(Duration::from_secs(2)),
+					..Default::default()
+				},
+			);
+		}
+
+		// Five seconds between frames, past the second a skip keeps before what comes next.
+		let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
+		for (sec, slice) in [(0, 0x65u8), (5, 0x41)] {
+			producer
+				.write(Frame {
+					timestamp: Timestamp::from_micros(sec * 1_000_000).unwrap(),
+					duration: None,
+					payload: length_prefixed(&[&[slice; 300]]),
+					keyframe: sec == 0,
+				})
+				.unwrap();
+		}
+		producer.finish().unwrap();
+
+		// No delay: each frame settles the slots before it, so the gap is skipped at once.
+		let mut export = open_ts(
+			crate::source::announced(&consumer).await,
+			crate::catalog::CatalogFormat::Hang,
+		)
+		.await
+		.unwrap()
+		.with_replay();
+		if let Some(rate) = rate {
+			export = export.with_mux_rate(rate);
+		}
+		let mut ts = BytesMut::new();
+		while let Ok(frame) = tokio::time::timeout(DRAIN, export.next()).await {
+			let Some(frame) = frame.expect("exporter error") else {
+				break;
+			};
+			ts.extend_from_slice(&frame.payload);
+		}
+		assert_packet_aligned(&ts);
+
+		// The two PCRs between the PMT and the first SI step forward by at most a valid interval.
+		let packets = ts.as_chunks::<188>().0;
+		let pid = |packet: &[u8; 188]| u16::from(packet[1] & 0x1f) << 8 | u16::from(packet[2]);
+		let pmt = packets.iter().position(|packet| pid(packet) == 0x1000).expect("a PMT");
+		let first = packets.iter().position(|packet| pid(packet) == 0x0011).expect("SI");
+		let pcrs: Vec<u64> = packets[pmt..first].iter().filter_map(super::import::pcr).collect();
+		let [.., before, last] = pcrs[..] else {
+			panic!("{rate:?}: {} PCRs between the PMT and the first SI", pcrs.len());
+		};
+		let step = (last + super::mux_rate::PCR_WRAP - before) % super::mux_rate::PCR_WRAP;
+		assert!(
+			(1..=super::mux_rate::MAX_INTERVAL).contains(&step),
+			"{rate:?}: the PCR steps {before} to {last} before the first SI"
+		);
+
+		let mut broadcast2 = moq_net::broadcast::Info::new().produce();
+		let catalog2 = crate::catalog::Producer::new(
+			&mut broadcast2,
+			crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
+		)
+		.unwrap();
+		let mut import = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
+		import.decode(&BytesMut::from(ts.as_ref())).unwrap();
+		import.finish().unwrap();
+		for (pid, stream) in &import.stats().streams {
+			assert_eq!(
+				stream.unplaced, 0,
+				"{rate:?}: PID {pid:#x} dropped what it could not place"
+			);
+		}
+	}
+}
+
 /// Each SI PID must be re-emitted on its own interval, independently of the PSI cadence
 /// and of video keyframes. The fixtures are a fraction of a second long, so nothing else
 /// distinguishes a correct interval from "emitted once and never again": this builds a
diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs
index 1d982d9ab8a..1266d994e2c 100644
--- a/rs/moq-mux/src/container/ts/schedule.rs
+++ b/rs/moq-mux/src/container/ts/schedule.rs
@@ -637,9 +637,10 @@ impl Schedule {
 		self.opening = true;
 	}
 
-	/// Whether a unit pushed now could go out before the opening's PCR interval is out.
+	/// Whether a unit pushed now could go out before the opening's PCR interval is out, or
+	/// ahead of packets still held for it on the same PID.
 	pub fn clocking(&self) -> bool {
-		self.opening || self.next.is_some_and(|next| next < self.clocked)
+		self.opening || !self.unclocked.is_empty() || self.next.is_some_and(|next| next < self.clocked)
 	}
 
 	/// Hold `packets`, which a reader places on the program clock and which ride no unit of their
@@ -1008,7 +1009,12 @@ impl Schedule {
 			(None, Some(first)) => Some(first),
 			// Skip the empty slots of a long gap, keeping the second before what comes next.
 			(Some(next), first) => {
-				let target = first.or(known);
+				let mut target = first.or(known);
+				// The opening's PCR interval runs slot by slot to its gate, so what it holds goes
+				// out on a clock the reader has measured.
+				if self.opening || !self.unclocked.is_empty() {
+					target = target.map(|target| target.min(self.clocked));
+				}
 				Some(next.max(target.map_or(0, |target| target.saturating_sub(BACKFILL))))
 			}
 		}
@@ -1820,6 +1826,46 @@ mod tests {
 		}
 	}
 
+	/// A gap after the first frames skips no slot of the opening's PCR interval, so packets held
+	/// for it go out after consecutive PCRs rather than after a jump the reader cannot measure,
+	/// padded or not. Until they do, SI muxed after them is held behind them.
+	#[test]
+	fn clocked_packets_follow_the_opening_across_a_gap() {
+		for rate in [None, Some(RATE)] {
+			let mut schedule = Schedule::new(Duration::ZERO);
+			schedule.set_rate(rate);
+			schedule.set_tables(100);
+			schedule.set_clock(1);
+			schedule.set_buffer(1, VIDEO);
+			schedule.set_buffer(0x11, Buffer::SYSTEM);
+			schedule.open();
+			schedule.push_clocked(unit(0x11, 1));
+			schedule.lead(Tables {
+				packets: [unit(0, 1), unit(100, 1)].concat(),
+				clock: 1,
+				rate,
+				added: vec![1],
+			});
+			schedule.push(1, timing(10), unit(1, 1), true).unwrap();
+			// The next frame is ten seconds out, far past a skip.
+			let known = Some(400);
+			let mut indexes = Vec::new();
+			let mut held = None;
+			while let Some(slot) = schedule.next(known).unwrap() {
+				indexes.push(slot.index);
+				if slot.packets.as_chunks::<{ TsPacket::SIZE }>().0.iter().any(|packet| pid(packet) == 0x11) {
+					held = Some(slot.index);
+					break;
+				}
+				// SI muxed now goes behind what is held for the same PID, never ahead of it.
+				assert!(schedule.clocking(), "{rate:?}: after slot {}", slot.index);
+			}
+			assert_eq!(held, Some(OPENING), "{rate:?}: slots {indexes:?}");
+			assert_eq!(indexes, (0..=OPENING).collect::<Vec<_>>(), "{rate:?}");
+			assert!(!schedule.clocking());
+		}
+	}
+
 	/// The pointer byte goes just before placement, so it cannot cost a section another packet.
 	#[test]
 	fn placed_first_byte_accounts_for_the_section_pointer() {

From a5af3eb0152759660ae24e64d5ff143a3c670e57 Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Wed, 28 Oct 2026 09:04:04 -0700
Subject: [PATCH 6/8] style(moq-mux): format the gap test

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-mux/src/container/ts/schedule.rs | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs
index 1266d994e2c..e0e21f125f1 100644
--- a/rs/moq-mux/src/container/ts/schedule.rs
+++ b/rs/moq-mux/src/container/ts/schedule.rs
@@ -1853,7 +1853,13 @@ mod tests {
 			let mut held = None;
 			while let Some(slot) = schedule.next(known).unwrap() {
 				indexes.push(slot.index);
-				if slot.packets.as_chunks::<{ TsPacket::SIZE }>().0.iter().any(|packet| pid(packet) == 0x11) {
+				if slot
+					.packets
+					.as_chunks::<{ TsPacket::SIZE }>()
+					.0
+					.iter()
+					.any(|packet| pid(packet) == 0x11)
+				{
 					held = Some(slot.index);
 					break;
 				}

From 25e97be27a1234f35660e16e71d2685a50d9e51e Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Wed, 28 Oct 2026 09:16:06 -0700
Subject: [PATCH 7/8] docs(moq-mux): say why a cleared schedule keeps a pending
 opening

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-mux/src/container/ts/schedule.rs | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs
index e0e21f125f1..5e656c660d0 100644
--- a/rs/moq-mux/src/container/ts/schedule.rs
+++ b/rs/moq-mux/src/container/ts/schedule.rs
@@ -972,6 +972,8 @@ impl Schedule {
 		self.last = None;
 		self.ended = false;
 		self.lead = None;
+		// An opening still pending stays: its tables never went out, so the reader has no clock
+		// for the next timeline to continue.
 		self.clocked = 0;
 		self.unclocked.clear();
 	}

From 25ebd81dfb94e3ad02632d6d38e82106066f7d7f Mon Sep 17 00:00:00 2001
From: Luke Curley <kixelated@gmail.com>
Date: Wed, 28 Oct 2026 10:05:09 -0700
Subject: [PATCH 8/8] fix(moq-mux): hold later SI until the opening's held SI
 has gone out

The held SI rides a unit on a lane of its own, so emptying `unclocked` into it
ended the ordering barrier while the unit could still have packets to send: a
newer SI riding a media unit could overtake its tail. `clocking()` now holds
until that unit finishes, and SI held behind it is queued only then.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/moq-mux/src/container/ts/schedule.rs | 71 +++++++++++++++++++++++--
 1 file changed, 66 insertions(+), 5 deletions(-)

diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs
index 5e656c660d0..b091c3cb0a7 100644
--- a/rs/moq-mux/src/container/ts/schedule.rs
+++ b/rs/moq-mux/src/container/ts/schedule.rs
@@ -603,6 +603,9 @@ pub(super) struct Schedule {
 	/// Packets a reader places on the program clock, held until the opening's PCR interval is
 	/// out ([`Self::push_clocked`]).
 	unclocked: Vec<u8>,
+	/// The unit held packets last went out in, by `seq`: what is muxed after them waits until it
+	/// has finished, as it rides a lane of its own.
+	deferred: Option<u64>,
 }
 
 impl Schedule {
@@ -627,6 +630,7 @@ impl Schedule {
 			opening: false,
 			clocked: 0,
 			unclocked: Vec::new(),
+			deferred: None,
 		}
 	}
 
@@ -638,9 +642,18 @@ impl Schedule {
 	}
 
 	/// Whether a unit pushed now could go out before the opening's PCR interval is out, or
-	/// ahead of packets still held for it on the same PID.
+	/// ahead of packets still held for it.
 	pub fn clocking(&self) -> bool {
-		self.opening || !self.unclocked.is_empty() || self.next.is_some_and(|next| next < self.clocked)
+		self.opening
+			|| !self.unclocked.is_empty()
+			|| self.deferring()
+			|| self.next.is_some_and(|next| next < self.clocked)
+	}
+
+	/// Whether the unit held packets last went out in has some still to send.
+	fn deferring(&self) -> bool {
+		self.deferred
+			.is_some_and(|seq| self.units.iter().any(|unit| unit.seq == seq))
 	}
 
 	/// Hold `packets`, which a reader places on the program clock and which ride no unit of their
@@ -976,6 +989,7 @@ impl Schedule {
 		// for the next timeline to continue.
 		self.clocked = 0;
 		self.unclocked.clear();
+		self.deferred = None;
 	}
 
 	/// The first slot `unit` may go out in: [`ahead`] of its due slot, and never before the
@@ -1049,13 +1063,15 @@ impl Schedule {
 		if self.opening && self.next.is_none() {
 			self.clocked = index + OPENING;
 		}
-		if !self.opening && index >= self.clocked && !self.unclocked.is_empty() {
+		// Those held behind earlier ones wait for them to finish.
+		if !self.opening && index >= self.clocked && !self.unclocked.is_empty() && !self.deferring() {
 			let packets = std::mem::take(&mut self.unclocked);
 			let by = (index + 1) * PCR_INTERVAL.as_nanos();
 			// They follow no unit, so the next one pushed keeps its own start.
 			let last = self.last;
-			self.enqueue(pid(&packets), Timing { by, decode: by }, packets, None)
-				.late = true;
+			let unit = self.enqueue(pid(&packets), Timing { by, decode: by }, packets, None);
+			unit.late = true;
+			self.deferred = Some(unit.seq);
 			self.last = last;
 		}
 		self.next = Some(index + 1);
@@ -1874,6 +1890,51 @@ mod tests {
 		}
 	}
 
+	/// Packets held for the opening that take more than one slot to go out keep SI muxed after
+	/// them held behind them until the last has, padded or not.
+	#[test]
+	fn clocked_packets_stay_ahead_until_they_finish() {
+		for rate in [None, Some(RATE)] {
+			let mut schedule = Schedule::new(Duration::ZERO);
+			schedule.set_rate(rate);
+			schedule.set_tables(100);
+			schedule.set_clock(1);
+			schedule.set_buffer(1, VIDEO);
+			schedule.set_buffer(0x11, Buffer::SYSTEM);
+			schedule.open();
+			schedule.push_clocked(unit(0x11, 100));
+			schedule.lead(Tables {
+				packets: [unit(0, 1), unit(100, 1)].concat(),
+				clock: 1,
+				rate,
+				added: vec![1],
+			});
+			schedule.push(1, timing(10), unit(1, 3), true).unwrap();
+			let mut sent = 0;
+			let mut slots = Vec::new();
+			while let Some(slot) = schedule.next(None).unwrap() {
+				let packets = slot.packets.as_chunks::<{ TsPacket::SIZE }>().0;
+				let held = packets.iter().filter(|&packet| pid(packet) == 0x11).count();
+				if held > 0 {
+					slots.push(slot.index);
+				}
+				sent += held;
+				if sent < 100 {
+					assert!(
+						schedule.clocking(),
+						"{rate:?}: {sent} of 100 out by slot {}",
+						slot.index
+					);
+				}
+			}
+			assert_eq!(sent, 100, "{rate:?}");
+			assert!(!schedule.clocking(), "{rate:?}");
+			if rate.is_some() {
+				assert!(slots.len() > 1, "the held packets span slots {slots:?}");
+			}
+		}
+	}
+
 	/// The pointer byte goes just before placement, so it cannot cost a section another packet.
 	#[test]
 	fn placed_first_byte_accounts_for_the_section_pointer() {