diff --git a/doc/bin/cli.md b/doc/bin/cli.md index e6af71843ba..d2b843b866a 100644 --- a/doc/bin/cli.md +++ b/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 diff --git a/quest/m1/README.md b/quest/m1/README.md index ab753ae3837..d74700b4ac9 100644 --- a/quest/m1/README.md +++ b/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` diff --git a/quest/m1/export-ts-linger.md b/quest/m1/export-ts-linger.md deleted file mode 100644 index d0fd7a703c7..00000000000 --- a/quest/m1/export-ts-linger.md +++ /dev/null @@ -1,29 +0,0 @@ -# [S] An export failure is not the broadcast ending - -## Goal - -`moq export ts --linger` reports an export failure while the broadcast is -still live as a failure, instead of treating it as the broadcast ending and -waiting out the linger (`rs/moq-cli/src/subscribe.rs`). - -## Plan - -Reported by t0ms while grading #4645, who offered to do the CLI side. - -The error alone can't tell the cases apart: a publisher drop or SIGKILL -usually surfaces as a track `Err` too, and that must keep lingering (see the -comment above the linger in `run_ts`). Classify by the broadcast's state -instead. If it is still announced and live, the export failed: report it and -fail the run as before. If it closed or was replaced, it ended, clean or not, -and the linger starts. The track error can arrive before the broadcast's close -does, so the check has to settle that race rather than read the state once. - -A test drives each case: an export failure on a live broadcast fails the run, -a clean finish lingers, and a killed publisher still lingers and resumes when -the broadcast returns. - -Public API: none. Wire: none. - -## Related - -- [#4645 report](https://github.com/moq-dev/moq/pull/4645#issuecomment-6013366214) - the motivating `missed a decode deadline` failure exists only on #4645's branch diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index c45906c96cf..69eb2f08ccc 100644 --- a/rs/moq-cli/src/args.rs +++ b/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, diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index e145b6bc06b..7aac36eed84 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/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)] diff --git a/rs/moq-cli/tests/export.rs b/rs/moq-cli/tests/export.rs index 5363875b43a..d0c703ea67c 100644 --- a/rs/moq-cli/tests/export.rs +++ b/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>>; @@ -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;