diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 235a015c52d..961ce45d632 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -449,25 +449,27 @@ and writes its packets back out unchanged, from its first group on, each run between two PCRs `--delay` after the first one's time on the source clock. The output is exactly as conformant as the source, and two exporters emit the same bytes. A run that arrives after its deadline is dropped and counted, leaving a -continuity error, and the rest carry on. It takes neither `--linger`, +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. `export ts --linger 10s` waits that +A stdout export ends with the broadcast. `export ts --follow 10s` waits that long for the same publisher instance to come back, and carries on with the same -stream. A return that cannot be read yet, such as one still routed through a -dead session, keeps it waiting for the next announcement; if the linger runs -out first, it exits 1 with that error. A replacement (a restarted publisher, under a new +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 it as a full program switch: a new PMT from its catalog, with every PID flagging -the break. `--linger` also bounds a mid-stream stitch from the replacement -announcement through its first catalog snapshot; expiry exits 1. The default -`--linger 0s` follows only a replacement whose catalog is already available. -Without `--stitch` the export stays on the old publisher while it is +the break. A stitch waits for the replacement's first catalog snapshot until it +arrives or the replacement goes, keeping the old publisher playing meanwhile, +so even the default `--follow 0s` switches to a replacement that is still +starting. Without `--stitch` the export stays on the old publisher while it is up. Only lite-07 and moq-transport draft 17+ sessions carry epochs, so on the -default version every return is a replacement. 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 break, -so the other formats refuse `--linger`. +default version every return is a replacement. A failure other than the +publisher leaving, such as a codec TS cannot carry, exits 1 at once. Only `ts` +can mark a break, so the other formats refuse `--linger`, the released spelling +of `--follow`. ## Reports and feedback diff --git a/doc/bin/srt.md b/doc/bin/srt.md index 1e0069edfc4..d8f65e9b144 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -28,11 +28,9 @@ Each connection publishes under a fresh [epoch](/concept/moq-lite#publisher-epochs), so an encoder that reconnects while its stale connection is still open replaces it at once: viewers see the broadcast restart, and their next subscribe reaches the new feed. An -`export srt` stream ends with its broadcast; `--linger` and `--stitch` follow a +`export srt` stream ends with its broadcast; `--follow` and `--stitch` follow a return or a replacement on the same connection, as they do for -[`export ts`](/bin/cli#retention-and-latency). The linger also bounds a -mid-stream replacement's first catalog; expiry ends the export with an error. -With the default `--linger 0s`, that catalog must already be available. +[`export ts`](/bin/cli#retention-and-latency). A multi-program feed is refused unless `--program` picks one: `--program 2` imports program 2 alone, and `--program all` diff --git a/doc/setup/redundancy.md b/doc/setup/redundancy.md index 6e3c5961e8f..6e9279ae4fd 100644 --- a/doc/setup/redundancy.md +++ b/doc/setup/redundancy.md @@ -70,14 +70,13 @@ For MPEG-TS output, opt into following replacements: ```bash moq --connect https://edge.example.com/anon \ --connect-version moq-transport-17 --broadcast event.hang \ - export ts --stitch --linger 10s > event.ts + export ts --stitch --follow 10s > event.ts ``` `--stitch` follows a replacement as a program switch: its catalog supplies a new PMT and every PID flags the discontinuity. Starting a backup can cause -this switch too. `--linger 10s` bounds waiting after an end and waiting for a -replacement's first catalog; choose that window for your application. It does -not make the switch hitless. Without `--stitch`, the TS export stays on the +this switch too. `--follow 10s` bounds waiting for a publisher after an end; +choose that window for your application. It does not make the switch hitless. Without `--stitch`, the TS export stays on the old instance while it is up, then fails on a replacement. Other stdout export formats end with their broadcast; they do not follow a replacement. TS passthrough also cannot stitch. See [CLI retention and latency](/bin/cli#retention-and-latency). diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index 0e1ce03f1e7..f8b77132bc0 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -61,8 +61,7 @@ Rejected in planning: `close` collides with `AutoCloseable`, and no single verb fits every language's reserved names. - A follow policy enum folding `linger` and `stitch`: one name per meaning - ([TS export follows by cause](/quest/m1/broadcast-lifecycle/ts-follow.md) - renames the export wait `--follow`) fixes the confusion more cheaply. + (the export wait is `--follow`) fixes the confusion more cheaply. ## Required diff --git a/quest/m1/broadcast-lifecycle/README.md b/quest/m1/broadcast-lifecycle/README.md index 74c505b99a5..5f7ecbf94ce 100644 --- a/quest/m1/broadcast-lifecycle/README.md +++ b/quest/m1/broadcast-lifecycle/README.md @@ -47,7 +47,6 @@ timer on main at 1582fdc07d: ## Required - [Stateless broadcasts](/quest/m1/broadcast-lifecycle/stateless.md) - `Finish`, its linger, and finished-only supersession are gone from moq-net, `@moq/net`, moq-ffi, moq-c, and every binding -- [TS export follows by cause](/quest/m1/broadcast-lifecycle/ts-follow.md) - `export ts|srt --linger` becomes `--follow `, a replacement's catalog has no separate deadline, and a track failure is classified by its cause instead of a 1s grace - [Timer cleanups](/quest/m1/broadcast-lifecycle/timers.md) - `ConnectionProps.linger` is gone and stale linger references are fixed ## Related diff --git a/quest/m1/broadcast-lifecycle/ts-follow.md b/quest/m1/broadcast-lifecycle/ts-follow.md deleted file mode 100644 index 2f075546f16..00000000000 --- a/quest/m1/broadcast-lifecycle/ts-follow.md +++ /dev/null @@ -1,39 +0,0 @@ -# [M] TS export follows by cause - -## Goal - -`moq export ts --linger` and `moq export srt --linger` become `--follow -`: how long to wait for the publisher, or with `--stitch` a -replacement, to come back. The default 0 means don't follow. The Rust -`container::ts::Follower::with_linger` becomes `with_follow`, matching -moq-srt's `Subscribe::with_follow` and `Client::with_follow`. A replacement's first catalog -is awaited until it arrives or the replacement withdraws, with no deadline of -its own. A track failure is classified by its cause, not by waiting -`CLOSE_GRACE` (1s) to see whether the announcement also goes. - -## Plan - -Decided with the maintainer on 2026-10-23: - -- One knob, one meaning. Today `--linger` also bounds the stitch catalog - fetch (`rs/moq-mux/src/container/ts/follower.rs`, `Awaiting { deadline }`), - so `--stitch --linger 0` fails on a cold remote replacement. The - replacement's own lifetime is the bound instead. -- `CLOSE_GRACE` exists because the track error and the announcement removal - arrive on different streams. Find what tells "the publisher left" from "this - export failed" in the track's own end: the abort cause a relay gives a copy - whose upstream went away. If no cause distinguishes them today, add one at - the source. A wire error code change updates the drafts. -- The hidden `--linger` on the other container exports - (`rs/moq-cli/src/args.rs`) keeps refusing; point its message at `--follow` - only where following applies. -- Move the follower onto `moq_mux::follow::Instances` here (decided - 2026-10-24), as `moq play`, `moqsrc`, and `transcode` already do. `with_linger` waits from the - announcement through resolution and the first catalog, but `Instances` - yields `Start` only after resolution, so settle what `--follow` bounds - before moving the timer onto it. -- Docs: `doc/bin/cli.md`, plus a migration note for moq-cli, moq-mux, and - moq-srt in the Impact bullet. - -Public API: moq-mux rename (breaking); moq-cli flag rename. -Wire: none unless the failure cause needs a new code. diff --git a/rs/justfile b/rs/justfile index df314f95b21..59150fa1e46 100644 --- a/rs/justfile +++ b/rs/justfile @@ -334,7 +334,7 @@ quic-sim: send-depth: bash sh/rs/send-depth.sh -# Run moq-cli's `moq export` process tests, which feed a real relay and wait out lingers in real time. +# Run moq-cli's `moq export` process tests, which feed a real relay and wait out follows in real time. [positional-arguments] cli-export *args: just rs test -p moq-cli --test integration --run-ignored only -E 'test(/^export::/)' "$@" diff --git a/rs/moq-cli/MIGRATE.md b/rs/moq-cli/MIGRATE.md index c8153a06780..84620196579 100644 --- a/rs/moq-cli/MIGRATE.md +++ b/rs/moq-cli/MIGRATE.md @@ -79,13 +79,14 @@ Rename `--max-age` to `--max-delay` on those exports. Each frame is written that long after its decode time; the older staleness spellings are refused with `--delay` named. Rename `--max-age` to `--delay`. -### `moq export ts --linger` follows only the same publisher instance +### `moq export ts --linger` is `--follow`, and follows only the same publisher instance -A restarted publisher exits 1 unless `--stitch` follows it as a program switch. -`moq export srt` takes the same two flags. +Rename `export ts --linger` to `--follow`; the old spelling stops the run naming it. `export srt` takes the same `--follow`. +A restarted publisher exits 1 unless `--stitch` follows it as a program switch, and a stitched replacement's first catalog is awaited until it arrives or the replacement goes, so `--follow 0s` no longer refuses a cold one. +A failure other than the publisher leaving, such as a codec TS cannot carry, exits at once rather than after a 1 s grace. ```bash -moq export ts --linger 10s --stitch ... +moq export ts --follow 10s --stitch ... ``` ### `moq export fmp4` fixes its track set at the init segment diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index d9ee240b7aa..58914cb3c20 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -365,8 +365,8 @@ impl Invocation { } if let Some(stdout) = export.sink.stdout() { anyhow::ensure!( - stdout.linger.is_zero() || matches!(stdout.format, SubscribeFormat::Ts), - "--linger needs an output that can mark a restart, and only `export ts` can" + stdout.follow.is_zero() || matches!(stdout.format, SubscribeFormat::Ts), + "--linger is gone, and following a broadcast's return needs an output that can mark the restart: only `export ts --follow` can" ); if matches!(stdout.format, SubscribeFormat::H264 | SubscribeFormat::H265) { anyhow::ensure!( @@ -1090,7 +1090,7 @@ impl ExportSink { let container = |format, container: &Container| Stdout { format, max_delay: container.max_delay.into(), - linger: container.linger.into(), + follow: container.linger.into(), stitch: false, fragment_duration: None, mux_rate: None, @@ -1099,7 +1099,7 @@ impl ExportSink { let buffered = |format, args: &Buffered| Stdout { format, max_delay: args.delay.into(), - linger: args.linger.into(), + follow: args.linger.into(), stitch: false, fragment_duration: None, mux_rate: None, @@ -1117,7 +1117,7 @@ impl ExportSink { Self::Ts(args) => Stdout { format: SubscribeFormat::Ts, max_delay: args.delay.into(), - linger: args.linger.into(), + follow: args.follow.into(), stitch: args.stitch, fragment_duration: None, mux_rate: args.mux_rate, @@ -1137,7 +1137,9 @@ pub struct Stdout { /// The staleness budget, also the FLV/MKV release delay and how far behind the source `ts` /// runs (`--delay`). pub max_delay: Duration, - pub linger: Duration, + /// How long to wait for the broadcast to come back (`ts` only; the others refuse their + /// hidden `--linger`). + pub follow: Duration, /// Follow a replacement as a program switch (`ts` only). pub stitch: bool, pub fragment_duration: Option, @@ -1154,8 +1156,8 @@ pub struct Container { #[usage(long, default = "500ms")] pub max_delay: moq_tokio::cli::Duration, - /// Accepted only to refuse a nonzero value with a pointer to `ts`, the one format - /// that can mark where a returned broadcast restarts. + /// The released spelling of `export ts --follow`, accepted only to refuse a nonzero value: + /// only TS can mark where a returned broadcast restarts. #[usage(long, default = "0s", hide = true)] pub linger: moq_tokio::cli::Duration, @@ -1190,7 +1192,8 @@ pub struct Buffered { /// drops none as late, but still skips a stalled group once a newer one is ready. #[usage(long, default = "500ms")] pub delay: moq_tokio::cli::Duration, - /// Accepted only to refuse a nonzero value: only TS can mark a broadcast restart. + /// The released spelling of `export ts --follow`, accepted only to refuse a nonzero value: + /// only TS can mark a broadcast restart. #[usage(long, default = "0s", hide = true)] pub linger: moq_tokio::cli::Duration, /// An older spelling of the staleness setting. @@ -1238,12 +1241,12 @@ pub struct Transport { #[usage(long, default = "1s")] pub delay: moq_tokio::cli::Duration, - /// How long to wait for the same publisher instance to come back once it ends (e.g. `10s`), - /// then carry on with the same stream. A replacement exits 1 unless `--stitch` is passed. - /// An export that fails while the broadcast is still up exits without waiting. - /// Also bounds a stitched replacement through its first catalog; zero requires it already available. + /// How long to wait for the same publisher instance, or with `--stitch` a replacement, to + /// come back once the broadcast ends (e.g. `10s`), then carry on with the same stream. + /// Zero follows only an instance already serving. A replacement exits 1 unless `--stitch` is + /// passed. A failure other than the publisher leaving exits at once. #[usage(long, default = "0s")] - pub linger: moq_tokio::cli::Duration, + pub follow: moq_tokio::cli::Duration, /// Follow another publisher instance that replaces the broadcast, as a full program switch: /// a new PMT from its catalog, flagged as a break on every PID. @@ -1259,9 +1262,13 @@ pub struct Transport { /// Write the multiplex an `import ts --passthrough` published back out byte-identical, /// paced on its own PCR, `--delay` behind it. Packets that arrive later than that are /// dropped, leaving a continuity error. - #[usage(long, conflicts("--linger", "--stitch", "--mux-rate"))] + #[usage(long, conflicts("--follow", "--stitch", "--mux-rate"))] pub passthrough: bool, + /// The released spelling of [`Self::follow`]. + #[usage(long = "linger", hide = true)] + linger: Option, + /// The released spelling of [`Self::delay`]. #[usage(long = "max-age", hide = true)] max_age: Option, @@ -1281,6 +1288,9 @@ impl Transport { if self.latency_max.is_some() { found.changed("--latency-max", None, "--delay", BEHIND); } + if self.linger.is_some() { + found.flag("--linger", None, "--follow"); + } found } } @@ -1318,20 +1328,29 @@ mod tests { let _ = Stage::to_kdl(); } - /// Only TS can mark where a returned broadcast restarts, so only `export ts` may linger. + /// Only TS can mark where a returned broadcast restarts, so only `export ts` follows one. + /// The released `--linger` points there, as a rename on `ts` and a refusal elsewhere. #[test] - fn linger_is_ts_only() { - let parse = |format: &str, linger: &str| { - Invocation::try_parse_from(["moq", "--connect", "http://relay", "export", format, "--linger", linger]) - .unwrap() + fn follow_is_ts_only() { + let parse = |format: &str, flag: &str, value: &str| { + Invocation::try_parse_from(["moq", "--connect", "http://relay", "export", format, flag, value]) }; - assert!(parse("ts", "10s").validate().is_ok()); - for format in ["fmp4", "h264", "h265"] { - let err = parse(format, "10s").validate().unwrap_err().to_string(); - assert!(err.contains("--linger"), "{format}: {err}"); + assert!(parse("ts", "--follow", "10s").unwrap().validate().is_ok()); + let Err(err) = parse("ts", "--linger", "10s") else { + panic!("the released spelling must not start a run"); + }; + assert!(err.to_string().contains("--linger -> --follow"), "{err}"); + for format in ["fmp4", "mkv", "flv", "h264", "h265"] { + assert!(parse(format, "--follow", "10s").is_err(), "{format} takes no --follow"); + let err = parse(format, "--linger", "10s") + .unwrap() + .validate() + .unwrap_err() + .to_string(); + assert!(err.contains("export ts --follow"), "{format}: {err}"); assert!( - parse(format, "0s").validate().is_ok(), - "{format}: no linger is always fine" + parse(format, "--linger", "0s").unwrap().validate().is_ok(), + "{format}: a zero --linger is always fine" ); } } @@ -1536,7 +1555,7 @@ mod tests { let stdout = export.sink.stdout().unwrap(); assert!(stdout.passthrough); assert_eq!(stdout.max_delay, Duration::from_secs(2)); - for flags in [&["--stitch"][..], &["--linger", "10s"], &["--mux-rate", "20000000"]] { + for flags in [&["--stitch"][..], &["--follow", "10s"], &["--mux-rate", "20000000"]] { assert!(parse(flags).is_err(), "{flags:?} must not start a run"); } } @@ -1777,16 +1796,16 @@ mod tests { ); } - /// `export srt` follows its broadcast with the same `--linger` and `--stitch` as `export ts`. + /// `export srt` follows its broadcast with the same `--follow` and `--stitch` as `export ts`. #[test] - fn export_srt_takes_linger_and_stitch() { + fn export_srt_takes_follow_and_stitch() { let cli = Invocation::try_parse_from([ "moq", "export", "srt", "--listen", "[::]:9000", - "--linger", + "--follow", "10s", "--stitch", ]) @@ -1797,7 +1816,7 @@ mod tests { let ExportSink::Srt(args) = &export.sink else { panic!("an export srt stage"); }; - assert_eq!(Duration::from(args.linger), Duration::from_secs(10)); + assert_eq!(Duration::from(args.follow), Duration::from_secs(10)); assert!(args.stitch); assert!( Invocation::try_parse_from(["moq", "import", "srt", "--listen", "[::]:9000", "--stitch"]).is_err(), diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index a9bce29b152..2d500cd3991 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -559,6 +559,7 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res // anything bound a port or dialed out. let mut client = None; let mut connection = None; + let local = tokio::task::LocalSet::new(); let result = async { let Attached { bandwidth, @@ -613,20 +614,34 @@ async fn run_stages(moq: MoqSide, stages: Vec, net: Net) -> anyhow::Res } if locals.is_empty() { - drive(tasks, signals, shutdown).await + drive(&mut tasks, signals, shutdown).await } else { - let local = tokio::task::LocalSet::new(); supervise(&local, locals.into_iter().map(Publish::run), &mut tasks); - local.run_until(drive(tasks, signals, shutdown)).await + local.run_until(drive(&mut tasks, signals, shutdown)).await } } .await; // Even on a setup error. - detach(connection, client, &cluster).await; + stop(detach(connection, client, &cluster), &mut tasks, local).await; result } +/// Leave the MoQ network, then stop the stages. +/// +/// In this order an interrupted stage's open tracks end with the session that served them, +/// which a relay reports as a lost route and a reader follows. Stopping the stages first would +/// end them unfinished, which a reader takes for the publisher failing. +async fn stop( + leave: impl std::future::Future, + tasks: &mut JoinSet>, + local: tokio::task::LocalSet, +) { + leave.await; + tasks.shutdown().await; + drop(local); +} + /// Close everything [`spawn_moq`] attached. The process exits next, so the relay /// and cluster peers only hear we left if the closes go out now. The connection /// first delivers what it queued, such as the finished tracks at stdin EOF, since @@ -820,7 +835,7 @@ fn spawn_export( let args = SubscribeArgs { format: stdout.format, max_delay: stdout.max_delay, - linger: stdout.linger, + follow: stdout.follow, stitch: stdout.stitch, fragment_duration: stdout.fragment_duration, mux_rate: stdout.mux_rate, @@ -924,12 +939,14 @@ async fn run_stdout( } /// Run until an endpoint ends, keeping all tasks alive while a signalled listener drains. +/// +/// The tasks are left for the caller to shut down once the MoQ side is detached. async fn drive( - mut tasks: JoinSet>, + tasks: &mut JoinSet>, mut signals: Signals, shutdown: Option, ) -> anyhow::Result<()> { - let result = loop { + loop { tokio::select! { biased; () = signals.recv() => { @@ -946,9 +963,7 @@ async fn drive( None => break Ok(()), } } - }; - tasks.shutdown().await; - result + } } /// The listener / HTTP-serving endpoints bridge one named broadcast, so an @@ -1027,7 +1042,7 @@ mod tests { trigger, observer: observer.clone(), }; - let driving = tokio::spawn(drive(tasks, signals, Some(shutdown))); + let driving = tokio::spawn(async move { drive(&mut tasks, signals, Some(shutdown)).await }); observer.started().await; assert!(!driving.is_finished(), "drain ended under an established session"); drop(session); @@ -1283,7 +1298,7 @@ mod tests { supervise(&local, pipelines, &mut tasks); let err = local - .run_until(drive(tasks, Signals::new().unwrap(), None)) + .run_until(drive(&mut tasks, Signals::new().unwrap(), None)) .await .unwrap_err(); assert!(err.to_string().contains("pipeline panicked"), "{err}"); @@ -1300,9 +1315,82 @@ mod tests { supervise(&local, pipelines, &mut tasks); local - .run_until(drive(tasks, Signals::new().unwrap(), None)) + .run_until(drive(&mut tasks, Signals::new().unwrap(), None)) + .await + .unwrap(); + } + + /// An interrupted stage's open track reaches a remote reader as its session closing, which + /// a TS follower follows, not as a track dropped unfinished (`Stream(Internal)`). + #[tokio::test(start_paused = true)] + async fn an_interrupted_stage_ends_its_tracks_as_the_session_closing() { + let mut config = moq_net::transport::memory::Config::default(); + config.protocol = Some("moq-lite-06".into()); + let (client, server) = moq_net::transport::memory::pair(config); + let relay = moq_tokio::origin::spawn(); + let subscriber = relay.clone(); + let accepting = tokio::spawn(async move { + let (session, driver) = moq_net::Server::new() + .with_subscriber(subscriber) + .accept(moq_net::time::Instant::now(), server) + .await + .unwrap(); + tokio::spawn(moq_net::time::run(driver)); + session + }); + let publisher = moq_tokio::origin::spawn(); + let (session, driver) = moq_net::Client::new() + .with_publisher(publisher.consume()) + .connect(moq_net::time::Instant::now(), client) + .await + .unwrap(); + tokio::spawn(moq_net::time::run(driver)); + let _relayed = accepting.await.unwrap(); + + let broadcast = publisher.create_broadcast("demo").unwrap(); + broadcast.announce(Default::default()).unwrap(); + let info = moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI); + let track = broadcast.create_track("live", info).unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(moq_net::Timestamp::ZERO, b"live".as_ref()).unwrap(); + group.finish().unwrap(); + + let reader = relay.consume(); + reader.routed("demo").await.unwrap(); + let mut live = reader + .request_broadcast("demo", None) + .await + .unwrap() + .track("live") + .unwrap() + .subscribe(None) .await .unwrap(); + assert!(live.recv_group().await.unwrap().is_some()); + + let local = tokio::task::LocalSet::new(); + let mut tasks = JoinSet::new(); + let stage: Pipeline = Box::pin(async move { + let _open = (broadcast, track); + std::future::pending().await + }); + supervise(&local, [stage], &mut tasks); + stop( + async move { + let _ = session.close().await; + }, + &mut tasks, + local, + ) + .await; + + let end = loop { + match live.recv_group().await { + Ok(Some(_)) => continue, + end => break end.map(|_| ()), + } + }; + assert!(matches!(end, Err(moq_net::Error::Session(_))), "{end:?}"); } /// A stream bind failure fails startup before the caller can report READY, and diff --git a/rs/moq-cli/src/srt.rs b/rs/moq-cli/src/srt.rs index caa20cef428..a08f65e35d4 100644 --- a/rs/moq-cli/src/srt.rs +++ b/rs/moq-cli/src/srt.rs @@ -53,12 +53,12 @@ pub struct ExportArgs { #[usage(flatten)] pub endpoint: Args, - /// How long to wait for the same publisher instance to come back once it ends (e.g. `10s`), - /// then carry on with the same stream. A replacement ends the stream unless `--stitch` - /// is passed. - /// Also bounds a stitched replacement through its first catalog; zero requires it already available. + /// How long to wait for the same publisher instance, or with `--stitch` a replacement, to + /// come back once the broadcast ends (e.g. `10s`), then carry on with the same stream. + /// Zero follows only an instance already serving. A replacement ends the stream unless + /// `--stitch` is passed. #[usage(long, default = "0s")] - pub linger: moq_tokio::cli::Duration, + pub follow: moq_tokio::cli::Duration, /// Follow another publisher instance that replaces the broadcast, as a full program switch /// on the same SRT connection. @@ -146,7 +146,7 @@ pub async fn listen_export( args: ExportArgs, ) -> anyhow::Result<()> { let mut server = Server::bind(addr, Duration::from(args.endpoint.latency)).await?; - let (follow, stitch) = (args.linger.into(), args.stitch); + let (follow, stitch) = (args.follow.into(), args.stitch); tracing::info!(%addr, %name, "SRT listening (export)"); notify_ready(); @@ -208,7 +208,7 @@ pub async fn connect_export( let client = moq_srt::Client::new(addr, resource) .with_max_delay(args.endpoint.latency.into()) - .with_follow(args.linger.into()) + .with_follow(args.follow.into()) .with_stitch(args.stitch); Ok(client.publish(&origin, &name).await?) } diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index ee20cc95a8c..ffa3f2f7e24 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -181,8 +181,9 @@ pub struct SubscribeArgs { /// this far behind the source (`--delay`). pub max_delay: Duration, - /// How long to wait for the broadcast to come back after it ends (TS only). - pub linger: Duration, + /// How long to wait for the broadcast, or with `stitch` a replacement, to come back after + /// it ends (TS only). + pub follow: Duration, /// Follow another publisher instance replacing the broadcast, as a program switch (TS only). pub stitch: bool, @@ -421,7 +422,7 @@ impl Subscribe { // written as it comes. The path's announcements carry it across the broadcast's // returns, as they drive a player. let mut ts = moq_mux::container::ts::Follower::new(ts)? - .with_linger(self.args.linger) + .with_follow(self.args.follow) .with_stitch(self.args.stitch); // Reports a track that stops reaching the output while the rest keeps flowing, // the way `publish` reports one that stops arriving. diff --git a/rs/moq-cli/tests/integration/export.rs b/rs/moq-cli/tests/integration/export.rs index 2da6e2d0426..2e1400e4cb2 100644 --- a/rs/moq-cli/tests/integration/export.rs +++ b/rs/moq-cli/tests/integration/export.rs @@ -1,9 +1,9 @@ -//! `moq export ts --linger` and `--stitch` over a real relay: the export rides out the same +//! `moq export ts --follow` and `--stitch` over a real relay: the export rides out the same //! publisher instance leaving and coming back, switches to a replacement only when asked, and //! exits with the verdict of the broadcast's last end. The other formats fail once the route //! they read dies. //! -//! Each test spawns `moq` processes that feed clips and wait out lingers in real time, tens of +//! Each test spawns `moq` processes that feed clips and wait out follows in real time, tens of //! seconds between them, so they are ignored here and run nightly as `just rs cli-export`. #![cfg(unix)] @@ -123,9 +123,9 @@ fn moq(relay: &Relay, args: &[&str]) -> tokio::process::Command { command } -/// Start `moq export ts --linger `, collecting its output and diagnostics. -fn export(relay: &Relay, linger: &str) -> Export { - export_with(relay, &[], &["--linger", linger]) +/// Start `moq export ts --follow `, collecting its output and diagnostics. +fn export(relay: &Relay, follow: &str) -> Export { + export_with(relay, &[], &["--follow", follow]) } /// A container both `import` and `export` speak, with a clip of it. @@ -220,7 +220,7 @@ fn collect( received.lock().unwrap().push((Instant::now(), output.len())); } Some(fence) = requested.recv() => { - // The follower entered linger only after each old stdout write and flush. + // The follower started following only after each old stdout write and flush. // This task owns the reader, so no read can race the queued-byte count. let mut pending: libc::c_int = 0; // SAFETY: the pipe is live and pending points to a writable c_int. @@ -313,14 +313,14 @@ fn adaptation_flags(packet: &[u8]) -> Option { #[tokio::test] #[ignore = "real-time process test: just rs cli-export"] -async fn a_clean_finish_exits_zero_once_the_linger_expires() { +async fn a_clean_finish_exits_zero_once_the_follow_expires() { let relay = relay().await; let mut export = export(&relay, "1s"); let (mut publisher, feeding) = import(&relay); let stdin = feeding.await.unwrap(); export.output_past(0).await; - // The linger can't start before stdin closes, but it can before the publisher exits. + // The follow can't start before stdin closes, but it can before the publisher exits. let closed = Instant::now(); drop(stdin); assert!(wait(&mut publisher).await.success()); @@ -332,19 +332,21 @@ async fn a_clean_finish_exits_zero_once_the_linger_expires() { ); assert!( closed.elapsed() >= Duration::from_millis(900), - "the export waited out its linger" + "the export waited out its follow" ); } #[tokio::test] #[ignore = "real-time process test: just rs cli-export"] -async fn a_drop_exits_one_once_the_linger_expires() { +async fn a_drop_exits_one_once_the_follow_expires() { let relay = relay().await; let mut export = export(&relay, "1s"); let (publisher, feeding) = import(&relay); let stdin = feeding.await.unwrap(); export.output_past(0).await; + // The follow can't start before the interrupt. + let interrupted = Instant::now(); interrupt(publisher, stdin).await; let (status, errors) = export.wait().await; @@ -353,12 +355,16 @@ async fn a_drop_exits_one_once_the_linger_expires() { Some(1), "a drop exits 1, got {status}; exporter stderr: {errors}" ); + assert!( + interrupted.elapsed() >= Duration::from_millis(900), + "the export waited out its follow rather than failing on the drop; exporter stderr: {errors}" + ); } /// An export that fails on its own, with the broadcast still up, has no return to wait for. #[tokio::test] #[ignore = "real-time process test: just rs cli-export"] -async fn an_export_failure_with_the_broadcast_up_exits_without_lingering() { +async fn an_export_failure_with_the_broadcast_up_exits_without_following() { let relay = relay().await; let export = export(&relay, "20s"); @@ -416,8 +422,8 @@ async fn interrupted(export: &mut Export, publisher: (Child, JoinHandle::new(), "every PID flags the break"); } -/// One publisher reconnects within the linger, continuing its capture under the same epoch +/// One publisher reconnects within the follow, continuing its capture under the same epoch /// and program before delivering a clean final finish. #[tokio::test] #[ignore = "real-time process test: just rs cli-export"] -async fn the_same_instance_within_the_linger_continues() { +async fn the_same_instance_within_the_follow_continues() { let relay = relay().await; let epoch = moq_net::Epoch::mint(); let lite07 = ["--connect-version", LITE_07]; @@ -529,7 +535,7 @@ async fn the_same_instance_within_the_linger_continues() { .with_reconnect(false); let url: url::Url = relay.url.parse().unwrap(); let connection = client.connect(url.clone()).established().await.unwrap(); - let mut export = export_with(&relay, &lite07, &["--linger", "4s"]); + let mut export = export_with(&relay, &lite07, &["--follow", "4s"]); let output = export.output.clone(); // Continue one capture across a transport loss. Replaying its beginning in a fresh @@ -694,7 +700,7 @@ async fn a_standby_before_its_init_segment_leaves_the_fmp4_exports_running_on_li async fn an_old_publisher_that_stays_up_keeps_the_export() { let relay = relay().await; let lite07 = ["--connect-version", LITE_07]; - let mut export = export_with(&relay, &lite07, &["--linger", "10s"]); + let mut export = export_with(&relay, &lite07, &["--follow", "10s"]); let output = export.output.clone(); let (mut old, feeding) = import_with(&relay, &lite07); @@ -926,7 +932,7 @@ async fn redundant_failover(bufferbloat: bool) { export_with( &subscribers, &["--connect-version", LITE_07], - &["--stitch", "--linger", "10s", "--delay", "3s"], + &["--stitch", "--follow", "10s", "--delay", "3s"], ) }) .collect(); diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index d5a1f78a0fa..c58e9f65c05 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -264,7 +264,7 @@ async fn a_follower_keeps_hangz_after_catalog_eof() { drop((broadcast, track, catalog)); let mut follower = super::Follower::new(export) .unwrap() - .with_linger(Duration::from_secs(1)); + .with_follow(Duration::from_secs(1)); let (broadcast, track, mut catalog) = publish(); // Only catalog.json.z exists, so falling back to the default encoding cannot subscribe. tokio::time::timeout( @@ -9083,43 +9083,89 @@ fn finish((broadcast, mut catalog): (moq_net::broadcast::Producer, crate::catalo drop((broadcast, catalog)); } -/// A failure while the broadcast stays announced is the export's own: it fails after the -/// grace, without waiting out the linger. +/// A failure whose cause is not the publisher leaving is the export's own: it fails at once, +/// however long the follow. #[tokio::test(start_paused = true)] -async fn a_follower_fails_with_the_broadcast_up_after_the_grace() { - let origin = crate::source::produce_origin(); - let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); - let mut follower = follower(&origin).await.with_linger(Duration::from_secs(10)); - assert!( - tokio::time::timeout(Duration::from_secs(1), follower.next()) - .await - .is_err(), - "nothing to mux yet" - ); +async fn a_follower_fails_at_once_on_its_own_failure() { + let causes = [ + moq_net::Error::Cancel, + moq_net::Error::Stream(moq_net::StreamError::Internal), + moq_net::Error::MalformedTrack, + ]; + for cause in causes { + let origin = crate::source::produce_origin(); + let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let mut follower = follower(&origin).await.with_follow(Duration::from_secs(10)); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "nothing to mux yet" + ); - track.abort(moq_net::Error::Cancel); - let start = tokio::time::Instant::now(); - let end = follower.next().await; - assert!(end.is_err(), "the export's own failure fails the follower: {end:?}"); - assert_eq!(start.elapsed(), Duration::from_secs(1)); + track.abort(cause.clone()); + let start = tokio::time::Instant::now(); + let end = follower.next().await; + assert!( + matches!(&end, Err(crate::Error::Moq(err)) if err.to_string() == cause.to_string()), + "{cause}: {end:?}" + ); + assert_eq!(start.elapsed(), Duration::ZERO, "{cause}"); + } } -/// A replacement followed with stitching that never serves its catalog gives up at the -/// linger, rather than waiting on the catalog past it. +/// A track that lost its route, or whose session ended, says the publisher left even with the +/// broadcast still announced: the follower waits out the follow for a return, then fails with +/// that cause. moq-transport's PUBLISH_DONE has no code for a lost route, so a relay's copy +/// reads there as an unregistered one. #[tokio::test(start_paused = true)] -async fn a_follower_return_without_a_catalog_expires_with_the_linger() { +async fn a_follower_waits_out_the_follow_on_a_lost_route() { + let causes = [ + moq_net::Error::Unroutable, + moq_net::Error::Stream(moq_net::StreamError::Unroutable), + moq_net::Error::Session(moq_net::SessionError::Cancel), + moq_net::Error::Remote(0), + ]; + for cause in causes { + let origin = crate::source::produce_origin(); + let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "nothing to mux yet" + ); + + track.abort(cause.clone()); + let start = tokio::time::Instant::now(); + let end = follower.next().await; + assert!( + matches!(&end, Err(crate::Error::Moq(err)) if err.to_string() == cause.to_string()), + "{cause}: {end:?}" + ); + assert_eq!(start.elapsed(), follow, "{cause}"); + } +} + +/// A replacement followed with stitching is awaited until its catalog arrives, with no deadline +/// of its own: one whose catalog request is never answered keeps the follower waiting past the +/// follow, until it goes. +#[tokio::test(start_paused = true)] +async fn a_follower_awaits_a_replacement_catalog_past_the_follow() { let origin = crate::source::produce_origin(); let first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger).with_stitch(true); - let start = tokio::time::Instant::now(); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow).with_stitch(true); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); // Replaced, but the replacement's catalog request is never answered. @@ -9129,58 +9175,70 @@ async fn a_follower_return_without_a_catalog_expires_with_the_linger() { moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), ) .unwrap(); - let _unanswered = second.dynamic(); - let end = follower.next().await.unwrap(); - assert!(end.is_none(), "a return that never resumes is no return"); - assert_eq!(start.elapsed(), linger); + let unanswered = second.dynamic(); + assert!( + tokio::time::timeout(follow * 3, follower.next()).await.is_err(), + "awaiting the replacement's catalog past the follow" + ); + + drop((second, unanswered)); + let end = tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .expect("the replacement going ends the wait"); + assert!(matches!(end, Ok(None)), "the original clean end: {end:?}"); } /// A return of the same instance whose catalog subscription is answered but never delivers a -/// snapshot is no return either: the linger still bounds it. +/// snapshot is awaited past the follow, and once it fails too, it was no return: the export +/// ends as it did before it. #[tokio::test(start_paused = true)] -async fn a_follower_return_whose_catalog_never_snapshots_expires_with_the_linger() { +async fn a_follower_awaits_a_return_catalog_until_it_fails() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger); - let start = tokio::time::Instant::now(); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); let second = origin .publish("live", moq_net::origin::Route::default().with_epoch(epoch)) .unwrap(); - let _silent = second.create_track(hang::Catalog::DEFAULT_NAME, None).unwrap(); - let end = tokio::time::timeout(linger * 3, follower.next()) + let silent = second.create_track(hang::Catalog::DEFAULT_NAME, None).unwrap(); + assert!( + tokio::time::timeout(follow * 3, follower.next()).await.is_err(), + "awaiting the return's catalog past the follow" + ); + + silent.abort(moq_net::Error::Unroutable).unwrap(); + let end = tokio::time::timeout(Duration::from_secs(1), follower.next()) .await - .expect("the linger bounds the return") - .unwrap(); - assert!(end.is_none(), "a return that never resumes is no return"); - assert_eq!(start.elapsed(), linger); + .expect("the failed return ends the wait"); + assert!(matches!(end, Ok(None)), "the original clean end: {end:?}"); + drop(second); } /// A return that goes again before its catalog resolves is no return yet: the follower keeps -/// lingering, and with nothing else back it yields the original clean end at the deadline. +/// following, and with nothing else back it yields the original clean end at the deadline. #[tokio::test(start_paused = true)] -async fn a_follower_keeps_lingering_when_a_return_goes_before_it_resolves() { +async fn a_follower_keeps_following_when_a_return_goes_before_it_resolves() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); let start = tokio::time::Instant::now(); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); // Back, but its catalog request is never answered before it goes again. @@ -9196,11 +9254,11 @@ async fn a_follower_keeps_lingering_when_a_return_goes_before_it_resolves() { ); drop((second, unanswered)); - let end = tokio::time::timeout(linger * 3, follower.next()) + let end = tokio::time::timeout(follow * 3, follower.next()) .await - .expect("the linger bounds the wait"); + .expect("the follow bounds the wait"); assert!(matches!(end, Ok(None)), "the original clean end: {end:?}"); - assert_eq!(start.elapsed(), linger); + assert_eq!(start.elapsed(), follow); } /// Announce `epoch` at `live`, refusing every track request with `NotFound` until the @@ -9219,20 +9277,20 @@ fn publish_refusing(origin: &moq_net::origin::Producer, epoch: &moq_net::Epoch) } /// A replacement that refuses its catalog while still announced is no return: the follower -/// keeps lingering, and follows the next instance to serve. +/// keeps following, and follows the next instance to serve. #[tokio::test(start_paused = true)] -async fn a_follower_keeps_lingering_through_a_refused_return() { +async fn a_follower_keeps_following_through_a_refused_return() { let origin = crate::source::produce_origin(); let first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger).with_stitch(true); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow).with_stitch(true); let start = tokio::time::Instant::now(); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); // Replaced, but the replacement refuses its catalog request. @@ -9241,7 +9299,7 @@ async fn a_follower_keeps_lingering_through_a_refused_return() { tokio::time::timeout(Duration::from_secs(2), follower.next()) .await .is_err(), - "the refusal keeps it lingering" + "the refusal keeps it following" ); refusing.abort(); @@ -9257,18 +9315,18 @@ async fn a_follower_keeps_lingering_through_a_refused_return() { finish(second); let end = follower.next().await; assert!(matches!(end, Ok(None)), "the replacement's clean end: {end:?}"); - assert_eq!(start.elapsed(), Duration::from_secs(4) + linger); + assert_eq!(start.elapsed(), Duration::from_secs(4) + follow); } /// A return that stays announced but refuses its catalog fails the follower only once the -/// linger runs out, with that refusal rather than the original clean end. +/// follow runs out, with that refusal rather than the original clean end. #[tokio::test(start_paused = true)] -async fn a_follower_reports_a_refused_return_at_the_linger() { +async fn a_follower_reports_a_refused_return_at_the_follow() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); let start = tokio::time::Instant::now(); finish(first); @@ -9276,17 +9334,17 @@ async fn a_follower_reports_a_refused_return_at_the_linger() { tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); let _refusing = publish_refusing(&origin, &epoch); - let end = tokio::time::timeout(linger * 3, follower.next()) + let end = tokio::time::timeout(follow * 3, follower.next()) .await - .expect("the linger bounds the wait"); + .expect("the follow bounds the wait"); assert!( matches!(end, Err(crate::Error::Moq(moq_net::Error::NotFound))), - "the refusal surfaces at the linger: {end:?}" + "the refusal surfaces at the follow: {end:?}" ); - assert_eq!(start.elapsed(), linger); + assert_eq!(start.elapsed(), follow); } /// A stitch whose replacement goes before it resolves is no switch yet: the export stays on @@ -9299,7 +9357,7 @@ async fn a_follower_stitches_onto_the_next_replacement_when_one_goes_before_it_r let mut follower = follower(&origin) .await .with_stitch(true) - .with_linger(Duration::from_secs(10)); + .with_follow(Duration::from_secs(10)); // Replaced, but the replacement's catalog request is never answered before it goes. let second = origin.publish("live", route()).unwrap(); @@ -9331,21 +9389,22 @@ async fn a_follower_stitches_onto_the_next_replacement_when_one_goes_before_it_r } /// A stitch that fails to resolve while a return is still awaiting its catalog goes back to -/// that return, so its end starts the linger over rather than the old deadline cutting it off. +/// that return, which ending before its first snapshot never resumed: the export ends as it did +/// before it, at the original follow. #[tokio::test(start_paused = true)] async fn a_follower_keeps_a_return_awaiting_its_catalog_when_a_stitch_goes() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger).with_stitch(true); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow).with_stitch(true); let start = tokio::time::Instant::now(); finish(first); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "lingering" + "following" ); // Back, with a catalog yet to deliver its first snapshot. @@ -9376,18 +9435,18 @@ async fn a_follower_keeps_a_return_awaiting_its_catalog_when_a_stitch_goes() { finish(second); let end = follower.next().await; assert!(matches!(end, Ok(None)), "{end:?}"); - assert_eq!(start.elapsed(), Duration::from_secs(3) + linger); + assert_eq!(start.elapsed(), follow); } -/// A failure whose broadcast went and came back before the follower looked is no failure of -/// the export's own: the return is followed instead of failing at the grace. +/// A track that lost its route is the publisher leaving, so the return that follows is carried +/// on into. #[tokio::test(start_paused = true)] -async fn a_follower_follows_a_return_that_came_back_during_the_grace() { +async fn a_follower_follows_a_return_after_a_lost_route() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&epoch)); let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); - let mut follower = follower(&origin).await.with_linger(Duration::from_secs(10)); + let mut follower = follower(&origin).await.with_follow(Duration::from_secs(10)); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await @@ -9395,15 +9454,9 @@ async fn a_follower_follows_a_return_that_came_back_during_the_grace() { "nothing to mux yet" ); - track.abort(moq_net::Error::Cancel); - assert!( - tokio::time::timeout(Duration::from_millis(100), follower.next()) - .await - .is_err(), - "waiting out the grace" - ); - // The end and the return both arrive before the follower looks again. - drop((broadcast, catalog)); + track.abort(moq_net::Error::Unroutable); + catalog.abort(moq_net::Error::Unroutable); + drop(broadcast); let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&epoch)); let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); assert!( @@ -9412,6 +9465,7 @@ async fn a_follower_follows_a_return_that_came_back_during_the_grace() { .is_err(), "carried on into the return" ); + assert!(follower.export().cataloged(), "the return served its catalog"); track.finish().unwrap(); finish((broadcast, catalog)); @@ -9453,27 +9507,29 @@ async fn a_follower_built_after_its_route_went_follows_the_return() { finish(first); tokio::time::sleep(Duration::from_secs(1)).await; - let linger = Duration::from_secs(10); - let mut follower = super::Follower::new(export).unwrap().with_linger(linger); + let follow = Duration::from_secs(10); + let mut follower = super::Follower::new(export).unwrap().with_follow(follow); assert!( tokio::time::timeout(Duration::from_secs(2), follower.next()) .await .is_err(), - "lingering" + "following" ); - let second = publish_bare(&origin, Some(&epoch)); + let mut second = publish_bare(&origin, Some(&epoch)); + std::ops::DerefMut::deref_mut(&mut second.1.modify().unwrap()); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), "carried on" ); + assert!(follower.export().cataloged(), "the return served its catalog"); - // Carried on into the return, so its end starts the linger over. + // Carried on into the return, so its end starts the follow over. finish(second); let end = follower.next().await; assert!(matches!(end, Ok(None)), "{end:?}"); - assert_eq!(start.elapsed(), Duration::from_secs(4) + linger); + assert_eq!(start.elapsed(), Duration::from_secs(4) + follow); } /// The export's own instance winning the path back from a more specific route that went is no @@ -9523,12 +9579,12 @@ async fn a_follower_takes_its_own_instance_winning_back_for_no_replacement() { } /// Without stitching, another instance taking the path fails the follower once the export -/// ends, however long the linger. +/// ends, however long the follow. #[tokio::test(start_paused = true)] async fn a_follower_fails_on_a_replacement_without_stitch() { let origin = crate::source::produce_origin(); let first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_linger(Duration::from_secs(10)); + let mut follower = follower(&origin).await.with_follow(Duration::from_secs(10)); finish(first); let _second = publish_bare(&origin, Some(&moq_net::Epoch::mint())); @@ -9543,7 +9599,7 @@ async fn a_follower_fails_on_a_replacement_without_stitch() { async fn a_follower_takes_an_epochless_return_as_a_replacement() { let origin = crate::source::produce_origin(); let first = publish_bare(&origin, None); - let mut follower = follower(&origin).await.with_linger(Duration::from_secs(10)); + let mut follower = follower(&origin).await.with_follow(Duration::from_secs(10)); finish(first); let _second = publish_bare(&origin, None); @@ -9554,35 +9610,37 @@ async fn a_follower_takes_an_epochless_return_as_a_replacement() { assert!(matches!(err, crate::Error::Replaced(_)), "{err}"); } -/// The same instance coming back within the linger carries the export on: when it ends -/// again, the linger starts over from there. +/// The same instance coming back within the follow carries the export on: when it ends +/// again, the follow starts over from there. #[tokio::test(start_paused = true)] async fn a_follower_continues_on_the_same_instance_returning() { let origin = crate::source::produce_origin(); let epoch = moq_net::Epoch::mint(); let first = publish_bare(&origin, Some(&epoch)); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_linger(linger); + let follow = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_follow(follow); let start = tokio::time::Instant::now(); finish(first); assert!( tokio::time::timeout(Duration::from_secs(2), follower.next()) .await .is_err(), - "lingering" + "following" ); - let second = publish_bare(&origin, Some(&epoch)); + let mut second = publish_bare(&origin, Some(&epoch)); + std::ops::DerefMut::deref_mut(&mut second.1.modify().unwrap()); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), "carried on" ); + assert!(follower.export().cataloged(), "the return served its catalog"); finish(second); let end = follower.next().await.unwrap(); assert!(end.is_none()); - assert_eq!(start.elapsed(), Duration::from_secs(3) + linger); + assert_eq!(start.elapsed(), Duration::from_secs(3) + follow); } /// Pull frames from `follower` until it ends, or until nothing more comes within the drain. @@ -9618,7 +9676,7 @@ async fn a_follower_continues_through_a_same_epoch_handoff_after_a_gap() { .with_replay(); let mut follower = super::Follower::new(export) .unwrap() - .with_linger(Duration::from_secs(10)); + .with_follow(Duration::from_secs(10)); let start = tokio::time::Instant::now(); for ms in (0..200).step_by(20) { write_aac(&mut track, ms); @@ -9628,7 +9686,9 @@ async fn a_follower_continues_through_a_same_epoch_handoff_after_a_gap() { // Left unpolled across the handoff, and long enough for the exact route's retraction to // end the export's request before the prefix arrives. - drop((exact, catalog, track)); + track.abort(moq_net::Error::Unroutable); + catalog.abort(moq_net::Error::Unroutable); + drop(exact); tokio::time::sleep(Duration::from_secs(1)).await; let mut info = moq_net::broadcast::Info::new(); info.epoch = Some(epoch); @@ -9694,47 +9754,53 @@ async fn a_follower_ignores_a_covering_prefix_while_the_exact_route_serves() { assert!(matches!(end, Ok(None)), "no replacement: {end:?}"); } -/// Both the catalog subscription and its first snapshot share the restart's linger bound. +/// A mid-stream stitch awaits the replacement's catalog with no deadline of its own, so even a +/// zero follow switches to a cold replacement once its first snapshot arrives. #[tokio::test(start_paused = true)] -async fn a_midstream_stitch_catalog_expires_at_the_linger() { - for snapshot in [false, true] { - for linger in [Duration::ZERO, Duration::from_secs(10)] { - let origin = crate::source::produce_origin(); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let second = origin - .publish( - "live", - moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), - ) - .unwrap(); - let _unanswered = second.dynamic(); - let _silent = snapshot.then(|| second.create_track(hang::Catalog::DEFAULT_NAME, None).unwrap()); - let waiter = kio::Waiter::noop(); - if !linger.is_zero() { - assert!(follower.poll_next(&waiter).is_pending()); - tokio::time::advance(linger - Duration::from_millis(1)).await; - assert!(follower.poll_next(&waiter).is_pending()); - tokio::time::advance(Duration::from_millis(1)).await; - } - let std::task::Poll::Ready(Err(err)) = follower.poll_next(&waiter) else { - panic!("replacement catalog wait must fail at the linger, snapshot={snapshot}, linger={linger:?}"); - }; - assert!(err.to_string().contains("live"), "{err}"); - assert!(err.to_string().contains("catalog"), "{err}"); - } +async fn a_midstream_stitch_awaits_the_replacement_catalog() { + for follow in [Duration::ZERO, Duration::from_secs(10)] { + let origin = crate::source::produce_origin(); + let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let mut follower = follower(&origin).await.with_stitch(true).with_follow(follow); + let epoch = moq_net::Epoch::mint(); + let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); + assert!( + tokio::time::timeout(Duration::from_secs(60), follower.next()) + .await + .is_err(), + "awaiting the replacement's catalog, follow={follow:?}" + ); + assert!(!follower.export().cataloged(), "follow={follow:?}"); + + std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "nothing to mux, follow={follow:?}" + ); + assert!( + follower.export().cataloged(), + "the late snapshot switches, follow={follow:?}" + ); + assert_eq!(follower.export().instance(), Some(&epoch)); } } -/// Re-pricing a replacement while a stitch resolves it is the same instance: it neither -/// restarts the resolve nor extends the linger bound. +/// A mid-stream stitch keeps the old instance playing while the replacement's catalog request +/// goes unanswered, rather than stalling the program on it. #[tokio::test(start_paused = true)] -async fn a_midstream_stitch_bound_survives_route_updates() { +async fn a_midstream_stitch_keeps_the_old_instance_playing_until_it_switches() { let origin = crate::source::produce_origin(); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let linger = Duration::from_secs(10); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let start = tokio::time::Instant::now(); + let (mut broadcast, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + let export = open_export(crate::source::resolved(&origin, "live").await) + .await + .unwrap() + .with_delay(RECORDING_MAX_AGE) + .with_replay(); + let mut follower = super::Follower::new(export).unwrap().with_stitch(true); + let second = origin .publish( "live", @@ -9742,78 +9808,28 @@ async fn a_midstream_stitch_bound_survives_route_updates() { ) .unwrap(); let _unanswered = second.dynamic(); - let reprice = async { - let mut cost = 0u64; - loop { - tokio::time::sleep(Duration::from_secs(1)).await; - cost += 1; - second.announce(second.route().unwrap().with_cost(cost)).unwrap(); - } - }; - let end = tokio::time::timeout(linger * 3, async { - tokio::select! { - end = follower.next() => end, - () = reprice => unreachable!("re-pricing never stops"), - } - }) - .await - .expect("updates extended the stitch bound"); - let err = end.expect_err("the stitch fails at its linger"); - assert!(err.to_string().contains("catalog"), "{err}"); - assert_eq!(start.elapsed(), linger); -} - -/// A catalog ready before the bound completes the switch, including zero linger. -#[tokio::test(start_paused = true)] -async fn a_midstream_stitch_catalog_within_linger_clears_the_bound() { - for linger in [Duration::ZERO, Duration::from_secs(10)] { - let origin = crate::source::produce_origin(); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let epoch = moq_net::Epoch::mint(); - let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); - let waiter = kio::Waiter::noop(); - if !linger.is_zero() { - assert!(follower.poll_next(&waiter).is_pending()); - tokio::time::advance(linger - Duration::from_millis(1)).await; - } - std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); - // Resolve the local async serving path so zero linger sees an already available - // subscription and snapshot, rather than a request still parked on the driver. - let broadcast = origin - .consume() - .request_broadcast("live", Some(epoch.clone())) - .await - .unwrap(); - let mut available = crate::catalog::Consumer::<()>::new(&broadcast, crate::catalog::CatalogFormat::Hang) - .await - .unwrap(); - use crate::catalog::Stream; - kio::wait(|waiter| available.poll_next(waiter)).await.unwrap().unwrap(); - let next = follower.poll_next(&waiter); - assert!(next.is_pending(), "{next:?}, linger={linger:?}"); - assert!(follower.export().cataloged()); - assert_eq!(follower.export().instance(), Some(&epoch)); - tokio::time::advance(Duration::from_secs(20)).await; - assert!( - follower.poll_next(&waiter).is_pending(), - "the first snapshot clears the bound" - ); + for ms in (0..200).step_by(20) { + write_aac(&mut track, ms); } + let (frames, end) = drain_follower(&mut follower).await; + assert!(end.is_none(), "still exporting: {end:?}"); + assert_eq!(pes_count(&frames), 10, "the old instance kept playing"); } /// A stitch onto a replacement that goes before it resolves leaves the export awaiting its -/// current replacement's catalog, by that replacement's own deadline. +/// current replacement's catalog, which still completes the switch when it arrives. #[tokio::test(start_paused = true)] -async fn a_nested_stitch_keeps_the_pending_catalog_deadline() { +async fn a_nested_stitch_keeps_awaiting_the_pending_replacement() { let origin = crate::source::produce_origin(); - let linger = Duration::from_secs(10); let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let start = tokio::time::Instant::now(); + let mut follower = follower(&origin) + .await + .with_stitch(true) + .with_follow(Duration::from_secs(10)); - // Switched onto a replacement whose catalog never delivers a snapshot. - let _second = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + // Switched onto a replacement whose catalog has no snapshot yet. + let epoch = moq_net::Epoch::mint(); + let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); assert!( tokio::time::timeout(Duration::from_secs(2), follower.next()) .await @@ -9836,66 +9852,25 @@ async fn a_nested_stitch_keeps_the_pending_catalog_deadline() { "resolving the next replacement" ); drop((third, unanswered)); - - let end = tokio::time::timeout(linger * 3, follower.next()) - .await - .expect("the second replacement's deadline still bounds it"); - let err = end.expect_err("a replacement without a catalog fails"); - assert!(err.to_string().contains("catalog"), "{err}"); - assert_eq!(start.elapsed(), linger); -} - -/// A replacement restored after its own deadline, when the next one goes before it resolves, -/// fails even with its late first snapshot ready. -#[tokio::test(start_paused = true)] -async fn a_nested_stitch_restored_after_its_deadline_fails() { - let origin = crate::source::produce_origin(); - let linger = Duration::from_secs(10); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let start = tokio::time::Instant::now(); - - // Switched onto a replacement whose catalog has no snapshot yet, bounded at 10s. - let epoch = moq_net::Epoch::mint(); - let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); assert!( - tokio::time::timeout(Duration::from_secs(2), follower.next()) + tokio::time::timeout(Duration::from_secs(30), follower.next()) .await .is_err(), - "awaiting the replacement's catalog" + "awaiting the second replacement's catalog again" ); - // Replaced again at 2s, bounded at 12s, but that one's catalog request is never answered. - let third = origin - .publish( - "live", - moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), - ) - .unwrap(); - let unanswered = third.dynamic(); + std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); assert!( tokio::time::timeout(Duration::from_secs(1), follower.next()) .await .is_err(), - "resolving the next replacement" + "nothing to mux" ); - - // At 11s, the third goes and the second's first snapshot is ready, both past its deadline. - tokio::time::advance(linger + Duration::from_secs(1) - start.elapsed()).await; - drop((third, unanswered)); - std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); - // Deliver the snapshot through the local serving path, so it is ready on the next poll. - let broadcast = origin.consume().request_broadcast("live", Some(epoch)).await.unwrap(); - let mut available = crate::catalog::Consumer::<()>::new(&broadcast, crate::catalog::CatalogFormat::Hang) - .await - .unwrap(); - use crate::catalog::Stream; - kio::wait(|waiter| available.poll_next(waiter)).await.unwrap().unwrap(); - - let std::task::Poll::Ready(Err(err)) = follower.poll_next(&kio::Waiter::noop()) else { - panic!("a snapshot after the restored replacement's deadline must not count"); - }; - assert!(err.to_string().contains("catalog"), "{err}"); + assert!( + follower.export().cataloged(), + "the second replacement served its catalog" + ); + assert_eq!(follower.export().instance(), Some(&epoch)); } /// A replacement whose catalog closes before its first snapshot fails the stitch, rather than @@ -9907,7 +9882,7 @@ async fn a_midstream_stitch_catalog_closed_without_a_snapshot_fails() { let mut follower = follower(&origin) .await .with_stitch(true) - .with_linger(Duration::from_secs(10)); + .with_follow(Duration::from_secs(10)); let (_second, mut catalog) = publish_bare(&origin, Some(&moq_net::Epoch::mint())); assert!( @@ -9926,38 +9901,6 @@ async fn a_midstream_stitch_catalog_closed_without_a_snapshot_fails() { assert!(err.to_string().contains("catalog"), "{err}"); } -/// A snapshot that arrives after the deadline, before the follower polls again, is too late. -#[tokio::test(start_paused = true)] -async fn a_midstream_stitch_catalog_after_the_linger_fails() { - let origin = crate::source::produce_origin(); - let linger = Duration::from_secs(10); - let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); - let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); - let epoch = moq_net::Epoch::mint(); - let (_second, mut catalog) = publish_bare(&origin, Some(&epoch)); - let waiter = kio::Waiter::noop(); - assert!(follower.poll_next(&waiter).is_pending()); - - tokio::time::advance(linger).await; - std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); - // Deliver the snapshot through the local serving path, so it is ready on the next poll. - let broadcast = origin - .consume() - .request_broadcast("live", Some(epoch.clone())) - .await - .unwrap(); - let mut available = crate::catalog::Consumer::<()>::new(&broadcast, crate::catalog::CatalogFormat::Hang) - .await - .unwrap(); - use crate::catalog::Stream; - kio::wait(|waiter| available.poll_next(waiter)).await.unwrap().unwrap(); - - let std::task::Poll::Ready(Err(err)) = follower.poll_next(&waiter) else { - panic!("a snapshot after the linger must not count"); - }; - assert!(err.to_string().contains("catalog"), "{err}"); -} - /// The audio PTS in `frames`. fn audio_pts(frames: &[Frame]) -> Vec { let bytes: Vec = frames.iter().flat_map(|f| f.payload.iter().copied()).collect(); diff --git a/rs/moq-mux/src/container/ts/follower.rs b/rs/moq-mux/src/container/ts/follower.rs index f20a4f6fb97..f38b1d829fa 100644 --- a/rs/moq-mux/src/container/ts/follower.rs +++ b/rs/moq-mux/src/container/ts/follower.rs @@ -1,4 +1,4 @@ -//! [`Follower`]: an [`Export`] driven by its path's announcements, as a player is. +//! [`Follower`]: an [`Export`] driven by the publisher instances serving its path, as a player is. use std::future::Future; use std::pin::Pin; @@ -9,103 +9,95 @@ use moq_clock::Instant; use super::{Export, catalog}; use crate::container::Frame; - -/// How long an export failure waits for its broadcast to go before it counts as the export's own. -/// -/// A killed publisher's tracks can error just before its route is withdrawn, so the two need not -/// land together. -const CLOSE_GRACE: Duration = Duration::from_secs(1); - -/// The subscriptions carrying an export on into another broadcast. -type Resolved = crate::Result<(moq_net::broadcast::Consumer, crate::catalog::Consumer)>; +use crate::follow::{Event, Instances}; #[cfg(not(target_family = "wasm"))] -type Resolving = Pin> + Send>>; +type Subscribing = Pin>> + Send>>; #[cfg(target_family = "wasm")] -type Resolving = Pin>>>; +type Subscribing = Pin>>>>; -/// What the announcements say serves the exported path. -#[derive(Clone, Debug, PartialEq, Eq)] +/// What the instances say serves the exported path. enum Serving { - /// Nothing: every route went. - Gone, - /// The publisher instance being exported. + /// The instance being exported, with no change since the export carried on into it. Ours, - /// Another instance: a newer epoch, or any return of an epochless route. - Other(Option), + /// Nothing: the instance handed out last went. + Gone, + /// The instance being exported, handed out again since: it came back after a gap. + Return(moq_net::broadcast::Consumer), + /// Another instance: another epoch, or any return of an epochless route. + Other(moq_net::broadcast::Consumer), } -/// A first catalog snapshot the export awaits, since a subscribed catalog that never answers is -/// no return or switch. -struct Awaiting { - /// When the linger runs out. - deadline: Instant, - missing: Missing, +/// Subscribing to an instance's catalog, to carry the export on into it. +struct Subscription { + target: moq_net::broadcast::Consumer, + catalog: Subscribing, } -/// What a catalog still missing at the deadline means. +/// Why the export awaits the first catalog snapshot of the instance it carried on into. enum Missing { - /// A return that never resumed: the export ends as it did before it. - Return(crate::Result<()>), - /// A replacement switched to mid-stream that never served its catalog: a failure. + /// Carried on after the export ended with `end`: an instance that ends before it serves + /// leaves the export waiting for another until `deadline`. + Ended { end: crate::Result<()>, deadline: Instant }, + /// A replacement switched to mid-stream, which ending before it serves is a failure. Switch, } enum State { - /// Exporting, until the deadline when its catalog is yet to deliver a snapshot. - Running { awaiting: Option }, - /// The export ended with `end`, waiting for the announcements to say what follows. - Settling { + /// Exporting, while switching to a replacement when stitching one, and awaiting the + /// catalog's first snapshot after carrying on into an instance. + Running { + awaiting: Option, + switching: Option>, + }, + /// The export ended with `end`, waiting until `deadline` for an instance to carry on into. + Settling { end: crate::Result<()>, deadline: Instant }, + /// The export ended with `end`, and an instance serving the path is being subscribed. + Resuming { + subscription: Subscription, end: crate::Result<()>, - /// Until when a failure with the broadcast still announced waits for it to go. - grace: Option, - /// When the linger runs out. deadline: Instant, }, - /// Resolving the instance serving the path now, to carry the export on into it. - Following { - resolving: Resolving, - /// Bounds the resolution and then the catalog's first snapshot. - awaiting: Awaiting, - /// Where the export carries on, bound and all, if the instance goes before it resolves. - fallback: Box>, - }, /// Nothing follows. Done, } -/// Carries an [`Export`] across its broadcast's returns, following the path's announcements. +/// Carries an [`Export`] across its broadcast's returns, following the publisher instances +/// serving its path through [`Instances`]. /// -/// The same publisher instance (the same epoch) returning within the [linger](Self::with_linger) +/// The same publisher instance (the same epoch) returning within the [follow](Self::with_follow) /// continues the stream. Another instance taking the path fails with /// [`Error::Replaced`](crate::Error::Replaced), unless [stitching](Self::with_stitch), which /// switches the program to it at once. Subscriptions are sticky, so without stitching a -/// replaced publisher that stays up keeps the export until it ends. An export that fails -/// while its broadcast stays announced fails here too, without lingering. +/// replaced publisher that stays up keeps the export until it ends. +/// +/// A track failure is followed only when its cause says the publisher left: the route serving +/// it was lost (`Unroutable`, which a moq-lite relay resets a copy with once its upstream +/// session ends), the session carrying it ended, or a moq-transport peer ended it with a code +/// it does not register, since PUBLISH_DONE has none for a lost route. Any other failure is +/// the export's own and fails here at once. /// -/// Announcements alone decide whether the broadcast returned: a return that fails to resolve -/// waits for the next one, and the linger running out reports that failure. +/// An instance that comes back is awaited until its catalog serves a first snapshot, with no +/// deadline but its own end. A return that refuses its catalog waits for the next change, and +/// the follow running out reports that refusal. A refusal to resolve the path at all fails at +/// once, as it does for every [`Instances`] reader. /// /// An epochless route has no instance to match, so any return of one is a replacement. pub struct Follower { export: Export>, state: State, - announced: moq_net::announce::Follow, + instances: Instances, path: moq_net::PathOwned, - /// The epoch of the instance being exported. - epoch: Option, serving: Serving, - /// Every route went since the export resolved its broadcast. - ended: bool, - /// An announcement arrived since the follower was built. - started: bool, + /// Taking the instances on hand when built, which replay the route the export resolved. + priming: bool, /// The origin closed, so nothing more is announced. closed: bool, - /// Why the last return failed to resolve, reported instead of the end if the linger runs out. + /// Why the last return refused its catalog, reported instead of the end if the follow runs out. failed: Option, - /// A return failed to resolve: nothing is followed until the next announcement. + /// The instance serving now refused its catalog: nothing is followed until the next change. refused: bool, - linger: Duration, + follow: Duration, stitch: bool, timer: Option>>, } @@ -116,50 +108,41 @@ impl Follower { /// Fails when the export's origin can never cover its path. pub fn new(export: Export>) -> crate::Result { let source = export.source(); - let announced = source.origin().follow(source.path())?; - let path = source.path().clone(); - let epoch = export.instance().cloned(); + let instances = Instances::new(source.origin(), source.path(), None)?; let mut follower = Self { + path: source.path().clone(), export, - state: State::Running { awaiting: None }, - announced, - path, - epoch, + state: State::Running { + awaiting: None, + switching: None, + }, + instances, serving: Serving::Ours, - ended: false, - started: false, + priming: true, closed: false, failed: None, refused: false, - linger: Duration::ZERO, + follow: Duration::ZERO, stitch: false, timer: None, }; - // Take the start naming the route the export resolved now, before anything else can - // fold into it. The cursor replays the routes on hand at once, so none means the route - // already went, and the next start is a return. - follower.poll_announced(&kio::Waiter::noop()); - if !follower.started { - follower.started = true; - follower.serving = Serving::Gone; - follower.ended = true; - } + // Take the instance on hand now, before a gap can fold into it: one handed out at once is + // the route the export resolved, so only a later one is a return. + follower.poll_instances(&kio::Waiter::noop())?; + follower.priming = false; Ok(follower) } - /// Wait up to `linger` for the same instance to return once the broadcast ends. Defaults to - /// [`Duration::ZERO`]. - /// - /// It bounds the whole return, and a mid-stream stitch from its restart to the - /// replacement catalog's first snapshot. Zero requires an already available catalog. - pub fn with_linger(mut self, linger: Duration) -> Self { - self.linger = linger; + /// Wait up to `follow` after the broadcast ends for the same instance, or with + /// [stitching](Self::with_stitch) a replacement, to come back. Defaults to + /// [`Duration::ZERO`], which carries on only into an instance already serving the path. + pub fn with_follow(mut self, follow: Duration) -> Self { + self.follow = follow; self } /// Follow another publisher instance taking the path, as a full program switch (see - /// [`Export::follow`]). Defaults to `false`. Its first catalog must arrive within - /// [`with_linger`](Self::with_linger), including when the old instance is still live. + /// [`Export::follow`]). Defaults to `false`. pub fn with_stitch(mut self, stitch: bool) -> Self { self.stitch = stitch; self @@ -185,322 +168,251 @@ impl Follower { } fn poll_state(&mut self, waiter: &kio::Waiter) -> Poll>> { - if let Some(expired) = self.expired() { - return Poll::Ready(expired); - } - loop { match std::mem::replace(&mut self.state, State::Done) { State::Done => return Poll::Ready(Ok(None)), - State::Running { mut awaiting } => { - self.poll_announced(waiter); - // Subscriptions are sticky, so a replacement leaves the export on the old - // instance until it ends, unless asked to switch at once. - if self.stitch && matches!(self.serving, Serving::Other(_)) { - tracing::info!(path = %self.path, "broadcast replaced, switching the program to the new instance"); - self.state = State::Following { - resolving: self.resolve(), - awaiting: Awaiting { - deadline: moq_clock::now() + self.linger, - missing: Missing::Switch, - }, - fallback: Box::new(State::Running { awaiting }), - }; - continue; + State::Running { + mut awaiting, + mut switching, + } => { + self.poll_instances(waiter)?; + if self.poll_switch(waiter, &mut switching)? { + awaiting = Some(Missing::Switch); } let next = self.export.poll_next(waiter); - if self.export.cataloged() { - awaiting = None; - } + awaiting = awaiting.filter(|_| !self.export.cataloged()); let end = match next { Poll::Ready(Ok(Some(frame))) => { - self.state = State::Running { awaiting }; + self.state = State::Running { awaiting, switching }; return Poll::Ready(Ok(Some(frame))); } - // A switch whose catalog closed before its first snapshot never served one. - Poll::Ready(Ok(None)) => match awaiting { - Some(Awaiting { - missing: Missing::Switch, - .. - }) => Err(self.missing_catalog()), - _ => Ok(()), - }, - Poll::Ready(Err(err)) => Err(err), + Poll::Ready(Err(err)) if !left(&err) => { + tracing::warn!(path = %self.path, %err, "export failed, so not following"); + return Poll::Ready(Err(err)); + } + Poll::Ready(end) => end.map(|_| ()), Poll::Pending => { - if let Some(awaiting) = awaiting.take_if(|awaiting| self.elapsed(waiter, awaiting.deadline)) - { - return Poll::Ready(self.expire(awaiting)); + // An instance that goes with its catalog still open never served. + if let Some(missing) = awaiting.take_if(|_| matches!(self.serving, Serving::Gone)) { + self.state = self.withdrawn(missing); + continue; } - self.state = State::Running { awaiting }; + self.state = State::Running { awaiting, switching }; return Poll::Pending; } }; - let now = moq_clock::now(); - // A failure ends the broadcast only if the broadcast goes too. - let grace = (end.is_err() && !self.ended && self.serving == Serving::Ours) - .then(|| now + CLOSE_GRACE.min(self.linger)); - if grace.is_none() { - self.lingering(&end); + // Nor did one whose catalog ended before its first snapshot. + if let Some(missing) = awaiting { + self.state = self.withdrawn(missing); + continue; } + self.following(&end); self.state = State::Settling { end, - grace, - deadline: now + self.linger, + deadline: moq_clock::now() + self.follow, }; } - State::Settling { - end, - mut grace, - deadline, - } => { - self.poll_announced(waiter); - // A broadcast that went, even one already back, was no failure of the export's own. - if grace.is_some() && (self.serving != Serving::Ours || self.ended) { - grace = None; - self.lingering(&end); - } - if let Some(at) = grace { - // One still announced cannot return, so the failure is the export's own. - if self.closed || self.elapsed(waiter, at) { - let err = end.expect_err("only a failure waits out the grace"); - tracing::warn!(path = %self.path, %err, "export failed with the broadcast still up, so not lingering"); - return Poll::Ready(Err(err)); - } - self.state = State::Settling { end, grace, deadline }; - return Poll::Pending; - } - - let follow = match &self.serving { - _ if self.refused => false, - Serving::Ours => self.ended, - Serving::Other(_) if self.stitch => true, + State::Settling { end, deadline } => { + self.poll_instances(waiter)?; + let target = match &self.serving { + _ if self.refused => None, + Serving::Return(target) => Some(target.clone()), + Serving::Other(target) if self.stitch => Some(target.clone()), Serving::Other(_) => return Poll::Ready(Err(crate::Error::Replaced(self.path.to_string()))), - Serving::Gone => false, + Serving::Ours | Serving::Gone => None, }; - if follow { - self.state = State::Following { - resolving: self.resolve(), - fallback: Box::new(State::Settling { - end: end.clone(), - grace: None, - deadline, - }), - awaiting: Awaiting { - deadline, - missing: Missing::Return(end), - }, + if let Some(target) = target { + self.state = State::Resuming { + subscription: self.subscribe(target), + end, + deadline, }; continue; } if self.closed || self.elapsed(waiter, deadline) { return Poll::Ready(self.gone(end)); } - self.state = State::Settling { end, grace, deadline }; + self.state = State::Settling { end, deadline }; return Poll::Pending; } - State::Following { - mut resolving, - awaiting, - fallback, + State::Resuming { + mut subscription, + end, + deadline, } => { - // Whatever the request waits on, the announcements say what serves the path - // now: rather than sit out a route that went, ask afresh from there. - if self.poll_announced(waiter) { - tracing::debug!(path = %self.path, "announcement while resolving, asking again"); - // A stitch still facing a replacement asks that one, within its original bound. - if matches!(*fallback, State::Running { .. }) && matches!(self.serving, Serving::Other(_)) { - self.state = State::Following { - resolving: self.resolve(), - awaiting, - fallback, - }; - continue; - } - // Otherwise it stays on the export it has, still awaiting what it was. - self.state = *fallback; - // The fallback's wait predates this poll, so it gets no same-poll grace. - if let Some(expired) = self.expired() { - return Poll::Ready(expired); - } + // Whatever the subscription waits on, the instances say what serves the path + // now: rather than sit out one that went, decide afresh from there. + if self.poll_instances(waiter)? { + tracing::debug!(path = %self.path, "instance changed while subscribing, deciding again"); + self.state = State::Settling { end, deadline }; continue; } - if let Poll::Ready(resolved) = waiter.poll_future(resolving.as_mut()) { - let (broadcast, catalog) = match resolved { - Ok(resolved) => resolved, - // A stitch from a live export fails loud: the replacement would not serve. - Err(err) if matches!(*fallback, State::Running { .. }) => return Poll::Ready(Err(err)), - // While lingering, a return that fails to resolve is no return yet: nothing - // serves the path until the next announcement says otherwise. The error is - // reported only if the linger runs out. - Err(err) => { - tracing::warn!(path = %self.path, %err, "broadcast failed to resolve, still waiting"); + match waiter.poll_future(subscription.catalog.as_mut()) { + Poll::Ready(Ok(catalog)) => { + self.carry_on(&subscription.target, catalog)?; + self.state = State::Running { + awaiting: Some(Missing::Ended { end, deadline }), + switching: None, + }; + } + // Refused, or gone before the instances say so: nothing is followed until + // the next change. Only the follow running out reports a refusal. + Poll::Ready(Err(err)) => { + tracing::warn!(path = %self.path, %err, "catalog subscription failed, still waiting"); + if !left(&err) { self.failed = Some(err); - self.refused = true; - self.state = *fallback; - if let Some(expired) = self.expired() { - return Poll::Ready(expired); - } - continue; } - }; - self.export.followed(&broadcast, catalog)?; - self.failed = None; - self.epoch = self.export.instance().cloned(); - self.serving = Serving::Ours; - self.ended = false; - tracing::info!( - path = %self.path, - epoch = self.epoch.as_ref().map(tracing::field::display), - "exporting broadcast" - ); - self.state = State::Running { - awaiting: Some(awaiting), - }; - continue; - } - if self.elapsed(waiter, awaiting.deadline) { - return Poll::Ready(self.expire(awaiting)); + self.refused = true; + self.state = State::Settling { end, deadline }; + } + Poll::Pending => { + self.state = State::Resuming { + subscription, + end, + deadline, + }; + return Poll::Pending; + } } - self.state = State::Following { - resolving, - awaiting, - fallback, - }; - return Poll::Pending; } } } } - /// Apply every announcement on hand, returning whether any changed what serves the path. - fn poll_announced(&mut self, waiter: &kio::Waiter) -> bool { + /// Drive a switch to the replacement serving the path when stitching, returning whether the + /// export carried on into it. + /// + /// The old instance keeps playing until the replacement's catalog is subscribed, however long + /// that takes, and a replacement that goes first is no switch. + fn poll_switch(&mut self, waiter: &kio::Waiter, switching: &mut Option>) -> crate::Result { + let target = match &self.serving { + Serving::Other(target) if self.stitch && !self.refused => target, + _ => { + *switching = None; + return Ok(false); + } + }; + if !switching + .as_ref() + .is_some_and(|switching| switching.target.is_clone(target)) + { + tracing::info!(path = %self.path, "broadcast replaced, switching the program to the new instance"); + *switching = Some(self.subscribe(target.clone())); + } + let subscription = switching.as_mut().expect("just set"); + match waiter.poll_future(subscription.catalog.as_mut()) { + Poll::Ready(Ok(catalog)) => { + let subscription = switching.take().expect("just polled"); + self.carry_on(&subscription.target, catalog)?; + Ok(true) + } + // Gone before the instances say so: no switch until the next change. + Poll::Ready(Err(err)) if left(&err) => { + tracing::warn!(path = %self.path, %err, "replacement went before its catalog subscribed"); + *switching = None; + self.refused = true; + Ok(false) + } + // The replacement would not serve, so the switch fails loud. + Poll::Ready(Err(err)) => Err(err), + Poll::Pending => Ok(false), + } + } + + /// Carry the export on into `target`, whose `catalog` is subscribed. + fn carry_on( + &mut self, + target: &moq_net::broadcast::Consumer, + catalog: crate::catalog::Consumer, + ) -> crate::Result<()> { + self.export.followed(target, catalog)?; + self.serving = Serving::Ours; + self.failed = None; + tracing::info!( + path = %self.path, + epoch = self.export.instance().map(tracing::field::display), + "exporting broadcast" + ); + Ok(()) + } + + /// Apply every change on hand, returning whether any came through. + fn poll_instances(&mut self, waiter: &kio::Waiter) -> crate::Result { let mut any = false; while !self.closed { - any |= match self.announced.poll_next(waiter) { - Poll::Ready(Some(event)) => self.apply(event), - Poll::Ready(None) => { - self.closed = true; - true + match self.instances.poll_next(waiter) { + Poll::Ready(Ok(Event::Start(broadcast))) => { + self.serving = match self.export.instance() { + // An epochless route has no instance to match, so only the broadcast + // being exported is its own. + None if broadcast.is_clone(&self.export.source().broadcast()) => Serving::Ours, + // Even the same broadcast handed out again is a return: the origin hands + // one out only for an announcement that brings it. + Some(epoch) if broadcast.info().epoch.as_ref() == Some(epoch) => match self.priming { + true => Serving::Ours, + false => Serving::Return(broadcast), + }, + _ => Serving::Other(broadcast), + }; } + Poll::Ready(Ok(Event::End)) => self.serving = Serving::Gone, + Poll::Ready(Err(crate::Error::Moq(moq_net::Error::Closed))) => self.closed = true, + Poll::Ready(Err(err)) => return Err(err), Poll::Pending => break, - }; + } + any = true; } self.refused &= !any; - any + Ok(any) } - /// Apply an announcement, returning whether it changed what serves the path. - fn apply(&mut self, event: moq_net::announce::Event) -> bool { - let first = !std::mem::replace(&mut self.started, true); - self.serving = match event { - moq_net::announce::Event::Start(announce) => { - let epoch = announce.route.epoch; - // The first start names the route the export resolved, as near as an epochless - // route can tell. - match epoch == self.epoch && (epoch.is_some() || first) { - true => Serving::Ours, - false => Serving::Other(epoch), - } - } - // The export's own instance winning the path back, as when a more specific route of - // another instance goes, is still its own. - moq_net::announce::Event::Restart(announce) => { - let epoch = announce.route.epoch; - match epoch.is_some() && epoch == self.epoch { - true => Serving::Ours, - false => Serving::Other(epoch), - } - } - moq_net::announce::Event::End(_) => { - self.ended = true; - Serving::Gone - } - // The same instance over another route, which subscriptions ride out. A gap that - // ended the export's request comes as an end and a start instead. - moq_net::announce::Event::Update(_) => return false, - }; - true + /// Begin subscribing to `target`'s catalog. + fn subscribe(&self, target: moq_net::broadcast::Consumer) -> Subscription { + let (broadcast, format) = (target.clone(), self.export.catalog_format()); + Subscription { + target, + catalog: Box::pin(async move { crate::catalog::Consumer::new(&broadcast, format).await }), + } } - /// Begin resolving the instance serving the path now. - fn resolve(&self) -> Resolving { - let epoch = match &self.serving { - Serving::Other(epoch) => epoch.clone(), - Serving::Ours | Serving::Gone => self.epoch.clone(), - }; - let request = self.export.source().origin().request_broadcast(&self.path, epoch); - let format = self.export.catalog_format(); - Box::pin(async move { - let broadcast = request.await?; - let catalog = crate::catalog::Consumer::new(&broadcast, format).await?; - Ok((broadcast, catalog)) - }) + /// Where the export goes once the instance it carried on into ends before its first snapshot. + fn withdrawn(&mut self, missing: Missing) -> State { + tracing::warn!(path = %self.path, "broadcast ended before serving its catalog"); + match missing { + Missing::Ended { end, deadline } => State::Settling { end, deadline }, + Missing::Switch => State::Settling { + end: Err( + anyhow::anyhow!("replacement broadcast `{}` ended before serving its catalog", self.path).into(), + ), + deadline: moq_clock::now() + self.follow, + }, + } } - fn lingering(&self, end: &crate::Result<()>) { - if self.linger.is_zero() { + fn following(&self, end: &crate::Result<()>) { + if self.follow.is_zero() { return; } match end { Ok(()) => { - tracing::info!(path = %self.path, linger = ?self.linger, "broadcast finished, waiting for it to return") + tracing::info!(path = %self.path, follow = ?self.follow, "broadcast finished, waiting for it to return") } Err(err) => { - tracing::warn!(path = %self.path, %err, linger = ?self.linger, "broadcast ended, waiting for it to return") + tracing::warn!(path = %self.path, %err, follow = ?self.follow, "broadcast ended, waiting for it to return") } } } - /// End a catalog wait already past its deadline, before a late snapshot can count. - /// - /// Only waits from an earlier poll are checked: within one poll a ready catalog still makes - /// it, which zero linger relies on. - fn expired(&mut self) -> Option>> { - let awaiting = match &self.state { - State::Running { awaiting } => awaiting.as_ref(), - State::Following { awaiting, .. } => Some(awaiting), - State::Settling { .. } | State::Done => None, - }; - if !awaiting.is_some_and(|awaiting| moq_clock::now() >= awaiting.deadline) { - return None; - } - let (State::Running { - awaiting: Some(awaiting), - } - | State::Following { awaiting, .. }) = std::mem::replace(&mut self.state, State::Done) - else { - unreachable!("only a running or following state awaits a catalog"); - }; - Some(self.expire(awaiting)) - } - - /// How a wait for the catalog's first snapshot ends once the linger runs out. - fn expire(&mut self, awaiting: Awaiting) -> crate::Result> { - match awaiting.missing { - Missing::Return(end) => self.gone(end), - Missing::Switch => Err(self.missing_catalog()), - } - } - - /// How the export ends once the linger runs out with nothing back: with why the last - /// return failed to resolve, if one did, else as the broadcast ended. + /// How the export ends once the follow runs out with nothing back: with why the last + /// return refused its catalog, if one did, else as the broadcast ended. fn gone(&mut self, end: crate::Result<()>) -> crate::Result> { - tracing::info!(path = %self.path, linger = ?self.linger, "broadcast did not return"); + tracing::info!(path = %self.path, follow = ?self.follow, "broadcast did not return"); match self.failed.take() { Some(err) => Err(err), None => end.map(|()| None), } } - fn missing_catalog(&self) -> crate::Error { - anyhow::anyhow!( - "replacement broadcast `{}` did not serve its catalog within {:?}", - self.path, - self.linger - ) - .into() - } - /// Whether `at` has come, registering `waiter` to wake then if not. fn elapsed(&mut self, waiter: &kio::Waiter, at: Instant) -> bool { if moq_clock::now() >= at { @@ -513,3 +425,21 @@ impl Follower { waiter.poll_future(timer.as_mut()).is_ready() } } + +/// Whether a failure says the publisher left rather than the export failing: the route serving +/// the broadcast was lost, the session carrying it ended, or a moq-transport peer named no cause. +fn left(err: &crate::Error) -> bool { + let (crate::Error::Moq(err) | crate::Error::Hang(hang::Error::Moq(err))) = err else { + return false; + }; + use moq_net::{Error, StreamError}; + matches!( + err, + Error::Unroutable + | Error::Stream(StreamError::Unroutable | StreamError::Session(_)) + | Error::Session(_) + | Error::Transport(_) + | Error::IdleTimeout + | Error::Remote(_) + ) +} diff --git a/rs/moq-rtc/src/ingest.rs b/rs/moq-rtc/src/ingest.rs index 86d1b8805aa..859f3703e31 100644 --- a/rs/moq-rtc/src/ingest.rs +++ b/rs/moq-rtc/src/ingest.rs @@ -69,6 +69,7 @@ impl session::MediaSink for IngestSink { } fn abort(&mut self, err: moq_net::Error) { + self.catalog.clone().abort(err.clone()); self.bridges.abort(err); } diff --git a/rs/moq-rtc/src/server/whip.rs b/rs/moq-rtc/src/server/whip.rs index 134c886dbfa..9e8d6ad78e1 100644 --- a/rs/moq-rtc/src/server/whip.rs +++ b/rs/moq-rtc/src/server/whip.rs @@ -214,6 +214,34 @@ mod tests { offer.to_sdp_string() } + /// A session that ends without a DELETE, here one that never connects, ends its tracks + /// with the transport cause, so a reader sees the publisher go rather than a producer + /// dropped unfinished. + #[tokio::test(start_paused = true)] + async fn a_session_ending_without_a_delete_ends_its_tracks() { + let server = Server::new(crate::server::Config::default()); + let origin = moq_tokio::origin::spawn(); + let consumer = origin.consume(); + let session = accept(&server, &origin, "live/cam0", &offer()).await.unwrap(); + let broadcast = consumer.request_broadcast("live/cam0", None).await.unwrap(); + let mut catalog = broadcast + .track(hang::Catalog::DEFAULT_NAME) + .unwrap() + .subscribe(None) + .await + .unwrap(); + + let ended = session.run().await; + assert!(matches!(ended, Err(crate::Error::IceTimeout)), "{ended:?}"); + let end = loop { + match catalog.recv_group().await { + Ok(Some(_)) => continue, + end => break end.map(|_| ()), + } + }; + assert!(matches!(end, Err(moq_net::Error::Transport(_))), "{end:?}"); + } + /// An encoder reconnecting to the same broadcast while its stale session is still /// open restarts the broadcast at once: each offer is its own epoch, so a fresh /// request reaches the reconnect, while the stale viewer stays on the old one until diff --git a/rs/moq-rtc/src/session.rs b/rs/moq-rtc/src/session.rs index 5788c109031..5307ae84fd7 100644 --- a/rs/moq-rtc/src/session.rs +++ b/rs/moq-rtc/src/session.rs @@ -45,8 +45,9 @@ pub trait MediaSink: Send { /// loop has already converted the timestamp to microseconds. fn on_frame(&mut self, mid: str0m::media::Mid, frame: codec::Frame) -> Result<()>; - /// Called once when the session ends with a genuine failure, so the sink can - /// abort its tracks with the real cause instead of a bare `Error::Dropped`. + /// Called once when the session ends other than by the peer's clean end, so the sink can + /// abort its tracks, the catalog included, with the real cause instead of a bare + /// `Error::Dropped`. fn abort(&mut self, err: moq_net::Error); /// Called once when the peer ends the session cleanly after its final media: finish @@ -154,11 +155,10 @@ impl Session { pub async fn run(&mut self) -> Result<()> { let result = self.run_loop().await; - // A genuine failure (not a normal peer disconnect or an unconnected offer) - // propagates to subscribers as the real cause; the normal ends just let the - // sink's tracks close as Dropped. + // Any end but the peer's DELETE, a disconnect or an offer that never connected + // included, ends the sink's tracks with its cause, so subscribers see the publisher + // go rather than its tracks dropped unfinished. if let Err(err) = &result - && !matches!(err, Error::SessionClosed | Error::IceTimeout) && let MediaRole::Ingest(sink) = &mut self.role { sink.abort(moq_net::Error::Transport(err.to_string())); diff --git a/rs/moq-rtmp/src/dial.rs b/rs/moq-rtmp/src/dial.rs index 3f659d0675b..03232084290 100644 --- a/rs/moq-rtmp/src/dial.rs +++ b/rs/moq-rtmp/src/dial.rs @@ -603,9 +603,10 @@ impl Publisher { Ok(moq_mux::Linger::new([self.broadcast.clone()], self, linger)) } - /// Flush any buffered media, then end the broadcast so the origin unannounces it at once. + /// Flush any buffered media and finish the catalog, then end the broadcast so the origin unannounces it at once. fn close(mut self) -> anyhow::Result<()> { self.importer.finish()?; + self.catalog.finish()?; self.broadcast.close(); Ok(()) } @@ -615,6 +616,7 @@ impl Publisher { /// /// Consumes the publisher and closes the broadcast. fn abort(self, err: moq_net::Error) { + self.catalog.abort(err.clone()); self.importer.abort(err); self.broadcast.close(); } diff --git a/rs/moq-rtmp/src/server.rs b/rs/moq-rtmp/src/server.rs index 42e8f951967..d30a6eb3e45 100644 --- a/rs/moq-rtmp/src/server.rs +++ b/rs/moq-rtmp/src/server.rs @@ -1536,10 +1536,11 @@ impl Publisher { Ok(moq_mux::Linger::new([self.broadcast.clone()], self, linger)) } - /// Flush any buffered media, close out the broadcast's open groups, and end + /// Flush any buffered media, finish the catalog, close out the broadcast's open groups, and end /// the broadcast so the origin unannounces it immediately. fn close(mut self) -> anyhow::Result<()> { self.importer.finish()?; + self.catalog.finish()?; self.broadcast.close(); Ok(()) } @@ -1550,6 +1551,7 @@ impl Publisher { /// /// Consumes the publisher and closes the broadcast. fn abort(self, err: moq_net::Error) { + self.catalog.abort(err.clone()); self.importer.abort(err); self.broadcast.close(); } @@ -1767,6 +1769,34 @@ mod tests { } } + /// A hangup ends the catalog too, so a reader sees the publisher go rather than a + /// producer dropped unfinished. + #[tokio::test(start_paused = true)] + async fn a_hangup_ends_the_catalog() { + let origin = moq_tokio::origin::spawn(); + let consumer = origin.consume(); + let (mut client, server) = tokio::io::duplex(64 * 1024); + let accept = tokio::spawn(accept_publish(server, origin.clone(), Duration::ZERO)); + let catalog = async { + consumer.routed("live/cam0").await.unwrap(); + let broadcast = consumer.request_broadcast("live/cam0", None).await.unwrap(); + let track = broadcast.track(hang::Catalog::DEFAULT_NAME).unwrap(); + track.subscribe(None).await.unwrap() + }; + // Holding the stream past the publish keeps the hangup until the catalog is subscribed. + let (mut catalog, ()) = tokio::join!(catalog, publish_one_keyframe(&mut client, Leave::Hangup)); + drop(client); + accept.await.unwrap(); + + let end = loop { + match catalog.recv_group().await { + Ok(Some(_)) => continue, + end => break end.map(|_| ()), + } + }; + assert!(end.is_ok(), "the catalog ends cleanly: {end:?}"); + } + /// A late subscriber still reads the finished media during the linger, and an encoder /// reconnecting under the same key takes the path without ending it. #[tokio::test(start_paused = true)] diff --git a/rs/moq-srt/src/dial.rs b/rs/moq-srt/src/dial.rs index f0fc8e0008a..cc84e22a2c5 100644 --- a/rs/moq-srt/src/dial.rs +++ b/rs/moq-srt/src/dial.rs @@ -134,10 +134,9 @@ impl Client { self } - /// Wait up to `follow` for the same publisher instance to come back once the broadcast - /// [`publish`](Self::publish) reads ends, carrying on with the same stream. Zero (the - /// default) ends the push at the broadcast's end. - /// A stitched replacement must also supply its first catalog within this bound. + /// Wait up to `follow` for the same publisher instance, or when stitching a replacement, to + /// come back once the broadcast [`publish`](Self::publish) reads ends, carrying on with the + /// same stream. Zero (the default) follows only an instance already serving. pub fn with_follow(mut self, follow: Duration) -> Self { self.follow = follow; self diff --git a/rs/moq-srt/src/server.rs b/rs/moq-srt/src/server.rs index af2c04e9570..8f04a759846 100644 --- a/rs/moq-srt/src/server.rs +++ b/rs/moq-srt/src/server.rs @@ -471,10 +471,9 @@ impl Subscribe { self.0.peer } - /// Wait up to `follow` for the same publisher instance to come back once the broadcast - /// ends, carrying on with the same stream. Zero (the default) closes the SRT stream at - /// the broadcast's end. - /// A stitched replacement must also supply its first catalog within this bound. + /// Wait up to `follow` for the same publisher instance, or when stitching a replacement, to + /// come back once the broadcast ends, carrying on with the same stream. Zero (the default) + /// follows only an instance already serving, else closes the SRT stream. pub fn with_follow(mut self, follow: Duration) -> Self { self.0.follow = follow; self diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index 360dc535d93..23fde702674 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -178,7 +178,7 @@ const MIN_DELAY: Duration = Duration::from_secs(1); pub(crate) struct Options { /// The muxer's delay, at least [`MIN_DELAY`] ([`Subscriber::new`]). pub latency: Duration, - /// How long a return or stitched replacement has to supply its first catalog. + /// How long to wait for the broadcast to come back once it ends. pub follow: Duration, /// Follow another instance replacing the broadcast, as a full program switch. pub stitch: bool, @@ -224,7 +224,7 @@ impl Subscriber { let catalog = source.catalog::(moq_mux::catalog::CatalogFormat::Hang).await?; let export = ts::Export::new(source, catalog).with_delay(options.latency.max(MIN_DELAY)); let follower = ts::Follower::new(export)? - .with_linger(options.follow) + .with_follow(options.follow) .with_stitch(options.stitch); Ok(Some(Self { follower })) }