fix(ts): hold the catalog until the initial mux rate is measured

doc/bin/cli.md

@@ -309,6 +309,8 @@ See [Authentication](/bin/relay/auth).
 fetchable, which the [HLS gateway](/bin/hls) depends on. `export --max-delay`
 (default 500 ms) is how far a stalled group may fall behind the live edge
 before *this* consumer skips it. Raising the first never delays playback.
+At stdin EOF, `import` keeps the finished broadcast up for its max-age, so a
+subscriber still catching up can drain it; Ctrl-C ends that early.
 
 `export ts` takes `--delay` (default 500 ms) instead, like an SRT receiver's
 latency: each frame is muxed that long after its decode time, all tracks in

quest/m1/README.md

@@ -192,6 +192,7 @@ session outcomes, and embedder refusals inside session outcomes.
 - [moq fetch closes cleanly](/quest/m1/moq-fetch-close.md) - `moq fetch` closes its session before exiting instead of leaving the relay to time it out
 - [Terminal dial errors](/quest/m1/moq-tokio-terminal-errors.md) - moq-tokio's reconnect loop stops on errors that can never succeed
 - [Plan: HLS export failure signal](/quest/m1/hls-failure-signal.md) - decide what players see when the exporter's timeline fails, live or history, so a truncated recording never looks complete
+- [Publisher linger](/quest/m1/publisher-linger.md) - every finishing publisher keeps its broadcast up for the media max-age so subscribers can drain it
 
 ## Related
 

quest/m1/publisher-linger.md

@@ -0,0 +1,31 @@
+# [M] A finishing publisher lingers for max-age
+
+## Goal
+
+Every publisher that ends a broadcast cleanly keeps it up for the media
+max-age afterwards, so a subscriber still catching up (one that read the
+final catalog late, or is draining a backlog) can request and drain the
+finished tracks. Cached content stays servable until it is stale.
+
+## Plan
+
+Decided (2026-10-11): a retention rule, not a race-hiding sleep. The bound is
+the tracks' max-age, never a new constant, and an interrupt (SIGINT/SIGTERM,
+or the library equivalent) ends it early.
+
+`moq import` from stdin already does this (#5287). Extend the rule
+to the other finishing publishers: the CLI's gateway imports (HLS, RTMP, SRT,
+RTC), `moq-gst` sinks, the FFI and its bindings, and `@moq/publish`.
+
+Decide where the rule lives. Recommended: the library, so a finished
+`broadcast::Producer` (or the session close that follows it) stays servable
+for its tracks' max-age without each publisher reimplementing it. Then
+`moq import` drops its own linger. Report the public API shape before
+landing it.
+
+Test with a mocked clock: a subscriber that requests a finished track within
+max-age gets its media, and the publisher closes once max-age passes.
+
+## Related
+
+- [One max_age meaning](/quest/m1/cache-max-age.md) - what max-age means for a superseded group

quest/m1/ts-live-clock.md

@@ -55,5 +55,5 @@ public API and wire changes are not expected.
 - [Passing full CI](https://github.com/moq-dev/moq/actions/runs/38178905198) - same-head contrast, including real live timing
 - [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](/quest/m1/tstd/mux-rate-hold.md) - #5287 owns initial rate publication and its finite-source behavior
+- [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/tstd/README.md

@@ -36,7 +36,6 @@ nightly.
 ## Required
 
 - [Send-ahead within the delay](/quest/m1/tstd/send-ahead.md) - total lag is `--delay`, send-ahead included, with a 1 s default
-- [Mux-rate hold](/quest/m1/tstd/mux-rate-hold.md) - import publishes its catalog once the mux rate is measured, so export is constant-rate from the start
 - [PSI runs](/quest/m1/tstd/psi-runs.md) - the PAT and PMT are deduplicated and paced so the system buffers never overflow
 
 ## Related

rs/moq-cli/src/args.rs

@@ -709,7 +709,8 @@ pub struct Import {
 	/// wait for a late group). The default suits a segmented egress (HLS/DASH), which may only
 	/// advertise segments that are still fetchable; lower it when nothing reads history and the
 	/// memory matters. Media tracks only -- the catalog and timeline are read at the live edge,
-	/// which is retained unconditionally.
+	/// which is retained unconditionally. A stdin import also stays up this long after EOF, so
+	/// a subscriber still catching up can drain the finished broadcast.
 	#[usage(long)]
 	pub max_age: Option<crate::duration::Duration>,
 

rs/moq-cli/src/main.rs

@@ -662,6 +662,8 @@ fn spawn_import(
 			let broadcast = origin.create_broadcast(&name).context("failed to create broadcast")?;
 			Publish::new(broadcast, &format, config)?
 		};
+		// Media stays servable at EOF until it goes stale, the same max-age relays retain it for.
+		let publish = publish.linger(max_age.unwrap_or(hang::container::MAX_AGE));
 		publish.announce(epoch)?;
 		local = Some(publish);
 	} else {

rs/moq-cli/src/publish.rs

@@ -290,6 +290,8 @@ pub struct Publish {
 	// tracks and importers don't. `None` for `import ts --program all`, whose
 	// programs each keep and announce their own.
 	broadcast: Option<moq_net::broadcast::Producer>,
+	// How long a finished stdin import stays up; see [`Self::linger`].
+	linger: std::time::Duration,
 }
 
 impl Publish {
@@ -314,6 +316,7 @@ impl Publish {
 					catalog: PublishCatalog::Ts(catalog),
 				},
 				broadcast: Some(broadcast),
+				linger: Default::default(),
 			});
 		}
 
@@ -354,6 +357,7 @@ impl Publish {
 				catalog: PublishCatalog::Media(catalog),
 			},
 			broadcast: Some(broadcast),
+			linger: Default::default(),
 		})
 	}
 
@@ -369,6 +373,7 @@ impl Publish {
 				catalog: PublishCatalog::TsPrograms,
 			},
 			broadcast: None,
+			linger: Default::default(),
 		}
 	}
 
@@ -407,9 +412,18 @@ impl Publish {
 				audio,
 			},
 			broadcast: Some(broadcast),
+			linger: Default::default(),
 		})
 	}
 
+	/// Keep a finished stdin import up for `linger` after EOF, so a subscriber that learns the
+	/// final catalog late can still request and drain its tracks. Pass the media max-age: the
+	/// content is servable until it goes stale. SIGINT or SIGTERM ends it early.
+	pub fn linger(mut self, linger: std::time::Duration) -> Self {
+		self.linger = linger;
+		self
+	}
+
 	/// Advertise the broadcast's path under `epoch`, now that the catalog tracks are in place.
 	pub fn announce(&self, epoch: moq_net::Epoch) -> anyhow::Result<()> {
 		let Some(broadcast) = &self.broadcast else {
@@ -423,7 +437,7 @@ impl Publish {
 	/// Drive the source until stdin EOF (or the capture devices stop).
 	pub async fn run(self) -> anyhow::Result<()> {
 		match self.source {
-			Source::Stream { decoder, catalog } => decode(decoder, catalog, tokio::io::stdin()).await,
+			Source::Stream { decoder, catalog } => decode(decoder, catalog, tokio::io::stdin(), self.linger).await,
 			#[cfg(feature = "capture")]
 			Source::Capture {
 				broadcast,
@@ -481,10 +495,12 @@ impl Publish {
 /// At EOF the media tracks finish, then the catalog does, while it still lists them: the
 /// renditions retire from the catalog as the decoder drops, which a subscriber would read
 /// as removed tracks, and a catalog dropped unfinished reads as a publisher that vanished.
+/// The finished tracks then stay servable for `linger`.
 async fn decode(
 	mut decoder: PublishDecoder,
 	mut catalog: PublishCatalog,
 	mut input: impl tokio::io::AsyncRead + Unpin,
+	linger: std::time::Duration,
 ) -> anyhow::Result<()> {
 	// An empty BytesMut grows to only 64 bytes and clearing it keeps that capacity.
 	// Stdin dispatches every read to a blocking thread, so those tiny reads can
@@ -529,7 +545,18 @@ async fn decode(
 		log.finish(&stats);
 	}
 	match outcome {
-		Ok(()) => catalog.finish(),
+		Ok(()) => {
+			catalog.finish()?;
+			// The decoder and catalog own the tracks, so holding them keeps the finished
+			// broadcast servable: a subscriber that saw the final catalog only at EOF still has
+			// to request its media.
+			tracing::info!(
+				?linger,
+				"input ended; serving the finished broadcast until its media is stale"
+			);
+			tokio::time::sleep(linger).await;
+			Ok(())
+		}
 		Err(err) => {
 			decoder.abort(moq_net::Error::Transport(err.to_string()));
 			Err(err)
@@ -1190,7 +1217,7 @@ pub(crate) mod tests {
 			remaining: BBB,
 			reads: 0,
 		};
-		decode(decoder, catalog, &mut input).await.unwrap();
+		decode(decoder, catalog, &mut input, Duration::ZERO).await.unwrap();
 		assert_eq!(
 			input.reads,
 			BBB.len().div_ceil(188) + 1,
@@ -1223,7 +1250,7 @@ pub(crate) mod tests {
 				panic!("expected a stream source");
 			};
 			let start = std::time::Instant::now();
-			decode(decoder, catalog, input).await.unwrap();
+			decode(decoder, catalog, input, Duration::ZERO).await.unwrap();
 			let elapsed = start.elapsed();
 			let bytes = INPUT.len() * copies;
 			println!(
@@ -1246,7 +1273,7 @@ pub(crate) mod tests {
 		let Source::Stream { decoder, catalog } = publish.source else {
 			panic!("expected a stream source");
 		};
-		decode(decoder, catalog, BBB).await.unwrap();
+		decode(decoder, catalog, BBB, Duration::ZERO).await.unwrap();
 		drop(publish.broadcast);
 
 		let mut last = None;
@@ -1264,6 +1291,51 @@ pub(crate) mod tests {
 		assert_eq!(last.audio.renditions.len(), 1, "the audio rendition is still listed");
 	}
 
+	/// After EOF the finished tracks stay servable for the linger, so a subscriber that reads
+	/// the final catalog only then still gets the media, and the import returns once it ends.
+	#[tokio::test(start_paused = true)]
+	async fn a_finished_import_serves_its_tracks_for_the_linger() {
+		const LINGER: Duration = Duration::from_secs(30);
+		let broadcast = moq_net::broadcast::Info::new().produce();
+		let consumer = broadcast.consume();
+		let publish = Publish::new(broadcast, &PublishFormat::Ts { program: None }, Default::default()).unwrap();
+		let mut catalogs = hang::catalog::Catalog::<()>::subscribe(&consumer).await.unwrap();
+
+		#[allow(irrefutable_let_patterns)]
+		let Source::Stream { decoder, catalog } = publish.source else {
+			panic!("expected a stream source");
+		};
+		let started = tokio::time::Instant::now();
+		let mut import = std::pin::pin!(decode(decoder, catalog, BBB, LINGER));
+		let name = tokio::select! {
+			res = &mut import => panic!("the import returned before its linger: {res:?}"),
+			name = async {
+				loop {
+					let catalog = catalogs.next().await.unwrap().expect("a catalog before the finish");
+					if let Some(name) = catalog.video.renditions.keys().next() {
+						break name.clone();
+					}
+				}
+			} => name,
+		};
+
+		// Subscribe only now, as a subscriber that learned the catalog at EOF does.
+		let track = consumer.track(&name).unwrap().subscribe(None);
+		let frame = tokio::select! {
+			res = &mut import => panic!("the import returned before its linger: {res:?}"),
+			frame = async {
+				let mut track = track.await.unwrap();
+				let mut group = track.recv_group().await.unwrap().expect("a video group");
+				group.read_frame().await.unwrap()
+			} => frame,
+		};
+		assert!(frame.is_some(), "the finished track still serves its media");
+
+		import.await.unwrap();
+		assert!(started.elapsed() >= LINGER, "returned after {:?}", started.elapsed());
+		drop(publish.broadcast);
+	}
+
 	/// A PAT listing two programs, then one MP2 PES of each: program 1 on PID `0x61` at 1 s
 	/// with fill bytes `0xAA`/`0xBB`, program 2 on PID `0x71` an hour later with `0xCC`/`0xDD`.
 	pub(crate) fn two_programs() -> Vec<u8> {

rs/moq-cli/tests/export.rs

@@ -111,7 +111,8 @@ fn import(relay: &Relay) -> (Child, JoinHandle<ChildStdin>) {
 /// [`import`] with `global` flags, such as the session version or the epoch.
 fn import_with(relay: &Relay, global: &[&str]) -> (Child, JoinHandle<ChildStdin>) {
 	let mut child = moq(relay, global)
-		.args(["import", "ts"])
+		// A finished import lingers for its max-age; keep that short so the publisher exits promptly.
+		.args(["import", "--max-age", "1s", "ts"])
 		.stdin(Stdio::piped())
 		.spawn()
 		.expect("spawn import");
@@ -218,7 +219,7 @@ async fn an_export_failure_with_the_broadcast_up_exits_without_lingering() {
 	let (mut export, _output) = export(&relay, "20s");
 
 	// TS cannot carry AV1, so the export fails on the catalog while the publisher stays up.
-	let mut publisher = moq(&relay, &["import", "fmp4"])
+	let mut publisher = moq(&relay, &["import", "--max-age", "1s", "fmp4"])
 		.stdin(Stdio::piped())
 		.spawn()
 		.expect("spawn import");