fix(moq-cli): an export failure on a live broadcast does not linger

doc/bin/cli.md

@@ -297,8 +297,9 @@ keyframe. The output's PCR follows the source's clock within what ISO/IEC
 
 A stdout export ends with the broadcast. `export ts --linger 10s` waits that
 long for a restarted publisher instead, and marks the break in the stream
-(PCR discontinuity, PAT/PMT re-sent). Only `ts` can mark a restart, so the
-other formats refuse `--linger`.
+(PCR discontinuity, PAT/PMT re-sent). An export that fails while the broadcast
+stays up, such as on a codec TS cannot carry, exits 1 without lingering. Only
+`ts` can mark a restart, so the other formats refuse `--linger`.
 
 ## Debugging
 

quest/m1/README.md

@@ -88,7 +88,6 @@ blocks. The quests that gated m0 lines moved under them.
 - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface
 - [Relay session limits](/quest/m1/relay-session-limits.md) - moq-relay sets per-session request limits, tighter for clients than peers, and the bindings name a refused request
 - [Churn with held subscriptions](/quest/m1/session-churn-held.md) - opening and closing a request costs the same with 1 or 1,024 held subscriptions
-- [Export linger](/quest/m1/export-ts-linger.md) - `moq export ts --linger` fails on an export error while the broadcast is live, and lingers once it closes or is replaced
 - [Archive](/quest/m1/archive/README.md) - record selected tracks to any object_store and replay them over FETCH or derived HLS; the catalog entry and format may break in place, since no archives exist
 - [In-band auth](/quest/m1/auth/README.md) - a session tells its peer what it may publish and subscribe to, unions tokens presented in band, and fails loud on an out-of-scope publish
 - [Dropped sources](/quest/m1/dropped-sources.md) - track consumers see the producer's real error on every end path, never `Dropped`

rs/moq-cli/src/args.rs

@@ -1020,6 +1020,7 @@ pub struct Transport {
 
 	/// How long to wait for the broadcast to come back once it ends (e.g. `10s`).
 	/// The output stops while it is gone and resumes flagged as a break.
+	/// An export that fails while the broadcast is still up exits without waiting.
 	#[usage(long, default = "0s")]
 	pub linger: crate::duration::Duration,
 

rs/moq-cli/src/subscribe.rs

@@ -394,11 +394,19 @@ impl Subscribe {
 
 			release(&ts.stats());
 
-			// Any end waits out the linger, and on expiry the last one is the result: a
+			// An end waits out the linger, and on expiry the last one is the result: a
 			// clean catalog finish exits 0, a drop or any other failure exits 1.
 			if linger.is_zero() {
 				return Ok(end?);
 			}
+			// A failure ends the broadcast only if the broadcast goes too. One that stays
+			// up cannot return, so the failure is the export's own and exits now.
+			if let Err(err) = &end
+				&& !closes_within(&broadcast, CLOSE_GRACE.min(linger)).await
+			{
+				tracing::warn!(%err, "export failed with the broadcast still up, so not lingering");
+				return Ok(end?);
+			}
 			match &end {
 				Ok(()) => tracing::info!(?linger, "broadcast finished, waiting for it to return"),
 				Err(err) => tracing::warn!(%err, ?linger, "broadcast ended, waiting for it to return"),
@@ -431,6 +439,17 @@ impl Subscribe {
 	}
 }
 
+/// How long an export failure waits for its broadcast to close before it counts as the export's own.
+///
+/// A killed publisher's tracks can error just before its broadcast closes, so the two need not
+/// land together.
+const CLOSE_GRACE: Duration = Duration::from_secs(1);
+
+/// Whether `broadcast` closes within `grace`.
+async fn closes_within(broadcast: &hang::moq_net::broadcast::Consumer, grace: Duration) -> bool {
+	tokio::time::timeout(grace, broadcast.closed()).await.is_ok()
+}
+
 /// Wait up to `linger` for the `ended` broadcast to return and `ts` to resume on it.
 ///
 /// The linger bounds the whole return, catalog subscription included: a returned
@@ -457,6 +476,37 @@ async fn resume_within(
 mod tests {
 	use super::*;
 
+	/// A broadcast that stays up is not closing, however long the grace.
+	#[tokio::test(start_paused = true)]
+	async fn a_live_broadcast_does_not_close_within_the_grace() {
+		let (origin, driver) = hang::moq_net::origin::Producer::new(Default::default());
+		tokio::spawn(hang::moq_net::time::run(driver));
+		let _live = origin.publish("live", Default::default()).unwrap();
+		let broadcast = origin.consume().request_broadcast("live", None).await.unwrap();
+
+		let start = tokio::time::Instant::now();
+		assert!(!closes_within(&broadcast, CLOSE_GRACE).await);
+		assert_eq!(start.elapsed(), CLOSE_GRACE);
+	}
+
+	/// A broadcast that closes just after its tracks fail is still an end.
+	#[tokio::test(start_paused = true)]
+	async fn a_close_just_after_the_failure_is_an_end() {
+		let (origin, driver) = hang::moq_net::origin::Producer::new(Default::default());
+		tokio::spawn(hang::moq_net::time::run(driver));
+		let live = origin.publish("live", Default::default()).unwrap();
+		let broadcast = origin.consume().request_broadcast("live", None).await.unwrap();
+
+		let gap = CLOSE_GRACE / 4;
+		tokio::spawn(async move {
+			tokio::time::sleep(gap).await;
+			drop(live);
+		});
+		let start = tokio::time::Instant::now();
+		assert!(closes_within(&broadcast, CLOSE_GRACE).await);
+		assert_eq!(start.elapsed(), gap);
+	}
+
 	/// A broadcast that returns but never serves its catalog gives up at the linger,
 	/// rather than waiting on the catalog past it.
 	#[tokio::test(start_paused = true)]

rs/moq-cli/tests/export.rs

@@ -13,6 +13,8 @@ use tokio::task::JoinHandle;
 const TIMEOUT: Duration = Duration::from_secs(30);
 /// About three seconds of HEVC and Opus.
 const CLIP: &[u8] = include_bytes!("../../moq-mux/src/container/ts/test_data/bbb_cbr.ts");
+/// Fragmented AV1, which the TS export does not carry.
+const AV1: &[u8] = include_bytes!("../../moq-mux/src/container/fmp4/test_data/av1.mp4");
 
 type Output = Arc<Mutex<Vec<u8>>>;
 
@@ -178,6 +180,33 @@ async fn a_drop_exits_one_once_the_linger_expires() {
 	assert_eq!(status.code(), Some(1), "a drop exits 1, got {status}");
 }
 
+/// An export that fails on its own, with the broadcast still up, has no return to wait for.
+#[tokio::test]
+async fn an_export_failure_with_the_broadcast_up_exits_without_lingering() {
+	let relay = relay().await;
+	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"])
+		.stdin(Stdio::piped())
+		.spawn()
+		.expect("spawn import");
+	let mut stdin = publisher.stdin.take().expect("stdin");
+	let published = Instant::now();
+	stdin.write_all(AV1).await.expect("write stdin");
+
+	let status = wait(&mut export).await;
+	assert_eq!(status.code(), Some(1), "an export failure exits 1, got {status}");
+	assert!(
+		published.elapsed() < Duration::from_secs(10),
+		"exited after {:?}, as though the broadcast had ended",
+		published.elapsed()
+	);
+
+	drop(stdin);
+	wait(&mut publisher).await;
+}
+
 #[tokio::test]
 async fn a_publisher_restarted_within_the_linger_resumes_the_output() {
 	let relay = relay().await;