test(mux): require export EOF after finishing timing fixtures

doc/bin/cli.md

@@ -526,11 +526,15 @@ continuity error, and the rest carry on. It takes neither `--follow`,
 `--stitch` nor `--mux-rate`. Only the hang catalog names the track, so an
 `.msf` broadcast needs `--catalog-format hang`.
 
-A stdout export ends with the broadcast. A publisher that leaves still gets
-what the export buffered of it out on schedule first. `export ts --follow 10s`
-then waits that long for the same publisher instance to come back, and carries
-on with the same stream. A return that refuses its catalog keeps it waiting
-for the next announcement; if the follow runs out first, it exits 1 with that
+A stdout export ends when its catalog ends and its media has drained, so
+finishing only the media tracks leaves it waiting for catalog updates.
+`export ts --passthrough` instead ends with its track, once the packets it
+buffered are out, even while the catalog stays open; only a catalog that ends
+without naming the track ends it first. A publisher that leaves still gets what
+the export buffered of it out on schedule first. `export ts --follow 10s` then
+waits that long for the same publisher instance to come back, and carries on
+with the same stream. A return that refuses its catalog keeps it waiting for the
+next announcement; if the follow runs out first, it exits 1 with that
 refusal. One that goes before serving, such as one still routed through a dead
 session, is no return. A replacement (a restarted publisher, under a new
 [epoch](/concept/moq-lite#publisher-epochs)) exits 1 unless `--stitch` follows

quest/m1/README.md

@@ -145,7 +145,6 @@ per-frame CPU growth since #5926.
 - [moq-c at parity](/quest/m1/c/README.md) - the hand-written `moq-c` ships at the cut with every m1 binding feature moq-ffi gains
 - [Bindings see a moq-transport run end as an end](/quest/m1/ffi-ietf-run-end.md) - an IETF publisher ending its run reaches moq-ffi consumers as an end, not `Internal`
 - [Byte-stream renditions form a ladder](/quest/m1/stream-ladder.md) - moq-ffi stream producers link as `lower` and disable unfundable rungs like frame producers
-- [moq export ends when the broadcast ends](/quest/m1/export-broadcast-end.md) - the export finishes on its own, proven without a give-up timeout
 - [A relay answers a SUBSCRIBE on a copy it kept after a fetch](/quest/m1/relay-fetched-copy-subscribe.md) - the downstream SUBSCRIBE_OK waits for the upstream edge instead of hanging
 - [A job's exact route continues its claim's readers](/quest/m1/announce-takeover.md) - a job's route inherits its claim's identity, replacing the ANNOUNCE_TAKEOVER predecessor design
 - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges with absolute or relative ends, in either order, and replaces FETCH; merged ranges never starve a subscriber; relays fill misses by range, including over moq-transport

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

@@ -106,7 +106,8 @@ fn sanitize_mux_rate(rate: u64) -> Option<u64> {
 /// 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 publisher that leaves instead fails the
+/// Returns `None` once the catalog ends and its media drains. Finishing only the media
+/// tracks keeps it waiting for catalog updates. A publisher that leaves instead fails the
 /// export with its cause once what was already buffered has gone out on schedule.
 ///
 /// A rendition the catalog moves to another track (a new `track` or `broadcast`) goes on out

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

@@ -325,7 +325,7 @@ impl Live {
 			.write(Frame {
 				timestamp: Timestamp::from_micros((index * AUDIO_US).strict_add_signed(self.offset)).unwrap(),
 				duration: None,
-				payload: Bytes::from_iter((0..180u16).map(|i| (i ^ index as u16) as u8)),
+				payload: audio_payload(index),
 				keyframe: index.is_multiple_of(AUDIO_GROUP),
 			})
 			.unwrap();
@@ -347,15 +347,21 @@ impl Live {
 
 	/// End the broadcast and let every leg render what it holds.
 	fn finish(&mut self, legs: &mut [&mut Leg]) {
+		self.finish_media();
+		self.catalog.finish().unwrap();
+		for leg in legs.iter_mut() {
+			leg.drain(&self.clock);
+		}
+	}
+
+	/// End the media tracks while the catalog can still announce more.
+	fn finish_media(&mut self) {
 		if let Some(mut group) = self.group.take() {
 			group.finish().unwrap();
 		}
 		self.stalled.clear();
 		self.video.finish().unwrap();
 		self.audio.finish().unwrap();
-		for leg in legs.iter_mut() {
-			leg.drain(&self.clock);
-		}
 	}
 
 	/// Run the clock to `at`, letting every leg read whenever it would have on its own.
@@ -370,7 +376,7 @@ impl Live {
 			let now = kio::clock::now();
 			for leg in legs.iter_mut() {
 				if leg.wake().is_some_and(|wake| wake <= now) {
-					leg.poll();
+					assert!(!leg.poll(), "export ended while the broadcast is live");
 					if !leg.every.is_zero() {
 						while leg.next_poll <= now {
 							leg.next_poll += leg.every;
@@ -426,7 +432,7 @@ impl Leg {
 	/// nothing woke finds nothing new.
 	fn poll_now(&mut self) {
 		if self.every.is_zero() && self.woken.0.load(Ordering::Relaxed) {
-			self.poll();
+			assert!(!self.poll(), "export ended while the broadcast is live");
 		}
 	}
 
@@ -443,21 +449,14 @@ impl Leg {
 		false
 	}
 
-	/// Read to the end of the broadcast, or until nothing is due and 10 ms pass with no end.
-	///
-	/// Only `clock` fires the export's timers, so the 10 ms is an advance of it: a timeout armed
-	/// on it would never fire.
+	/// Read to the end of the broadcast, advancing only to scheduled output.
 	fn drain(&mut self, clock: &kio::clock::Deadlines) {
 		while !self.poll() {
-			if let Some(deadline) = self.export.next_due() {
-				clock.advance(deadline.max(clock.now()));
-				continue;
-			}
-			clock.advance(clock.now() + Duration::from_millis(10));
-			if self.export.next_due().is_none() {
-				self.poll();
-				break;
-			}
+			let deadline = self
+				.export
+				.next_due()
+				.expect("export parked without ending or scheduled output");
+			clock.advance(deadline.max(clock.now()));
 		}
 	}
 
@@ -467,6 +466,65 @@ impl Leg {
 	}
 }
 
+/// Media drains while the catalog stays open; EOF needs both, in either finish order.
+#[tokio::test(start_paused = true)]
+async fn end_waits_for_catalog_and_media() {
+	for catalog_first in [false, true] {
+		let mut live = Live::new(1.0, 0).await;
+		let mut leg = live.join(Duration::ZERO).await;
+		live.run(2 * GOP, &mut [&mut leg], |_| false);
+		let expected = if catalog_first {
+			live.catalog.finish().unwrap();
+			// Catalog EOF fixes the layout, but the named tracks can still publish media.
+			live.run(3 * GOP, &mut [&mut leg], |_| false);
+			// Close the groups so their tails flush, leaving both tracks open for more.
+			live.group.take().unwrap().finish().unwrap();
+			live.audio.cut(None).unwrap();
+			3 * GOP as usize
+		} else {
+			live.finish_media();
+			2 * GOP as usize
+		};
+		loop {
+			assert!(
+				!leg.poll(),
+				"EOF before both catalog and media finish: catalog_first {catalog_first}"
+			);
+			let Some(deadline) = leg.export.next_due() else {
+				break;
+			};
+			live.clock.advance(deadline.max(live.clock.now()));
+		}
+		assert_eq!(
+			leg.units().iter().filter(|unit| unit.pid == VIDEO_PID).count(),
+			expected
+		);
+		let audio: Vec<Bytes> = (0..live.audio_index).map(audio_payload).collect();
+		assert_eq!(aac_frames(&leg), audio, "catalog_first {catalog_first}: audio drained");
+		if catalog_first {
+			live.finish_media();
+		} else {
+			assert!(!leg.woken.0.load(Ordering::Relaxed), "the exporter is parked");
+			live.catalog.finish().unwrap();
+			assert!(
+				leg.woken.0.load(Ordering::Relaxed),
+				"catalog EOF must wake the exporter"
+			);
+		}
+		leg.drain(&live.clock);
+		assert_eq!(leg.export.dropped(), 0);
+		assert_eq!(
+			leg.units().iter().filter(|unit| unit.pid == VIDEO_PID).count(),
+			expected
+		);
+		assert_eq!(
+			aac_frames(&leg),
+			audio,
+			"catalog_first {catalog_first}: no audio after EOF"
+		);
+	}
+}
+
 /// A PES start as a receiver sees it.
 #[derive(Debug, Clone, Copy)]
 struct Unit {
@@ -1182,6 +1240,45 @@ async fn live_audio_shares_its_pes() {
 	}
 }
 
+/// The raw AAC of the fixture's audio frame `index`.
+fn audio_payload(index: u64) -> Bytes {
+	Bytes::from_iter((0..180u16).map(|i| (i ^ index as u16) as u8))
+}
+
+/// Each AAC frame a leg rendered, in output order, with its ADTS header removed.
+fn aac_frames(leg: &Leg) -> Vec<Bytes> {
+	let bytes: Vec<u8> = leg
+		.out
+		.iter()
+		.flat_map(|(_, frame)| frame.payload.iter().copied())
+		.collect();
+	let mut reader = TsPacketReader::new(Cursor::new(bytes));
+	let mut pes: Vec<Vec<u8>> = Vec::new();
+	while let Some(packet) = reader.read_ts_packet().unwrap() {
+		if packet.header.pid.as_u16() != AUDIO_PID {
+			continue;
+		}
+		match packet.payload {
+			Some(TsPayload::PesStart(start)) => pes.push(start.data.to_vec()),
+			Some(TsPayload::PesContinuation(rest)) => {
+				pes.last_mut().expect("a PES start first").extend_from_slice(&rest)
+			}
+			_ => {}
+		}
+	}
+	let mut frames = Vec::new();
+	for data in &pes {
+		// One PES carries several frames.
+		let mut data = &data[..];
+		while !data.is_empty() {
+			let header = super::adts::Header::parse(data).unwrap();
+			frames.push(Bytes::copy_from_slice(&data[header.header_len..header.frame_len]));
+			data = &data[header.frame_len..];
+		}
+	}
+	frames
+}
+
 /// Each video frame's DTS (else PTS) by its PTS, in 90 kHz ticks, as a leg rendered it.
 fn decode_times(leg: &Leg) -> std::collections::BTreeMap<u64, u64> {
 	let bytes: Vec<u8> = leg