diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 652af7df57..f553d53c70 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -313,10 +313,15 @@ keyframe. The output's PCR follows the source's clock within what ISO/IEC 13818-1 allows. `--delay 0` writes frames in arrival order and drops nothing. A stdout export ends with the broadcast. `export ts --linger 10s` waits that -long for a restarted publisher instead, and marks the break in the stream -(PCR discontinuity, PAT/PMT re-sent). An export that fails while the broadcast -stays up, such as on a codec TS cannot carry, exits 1 without lingering. Only -`ts` can mark a restart, so the other formats refuse `--linger`. +long for the same publisher instance to come back, and carries on with the same +stream. 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. Without `--stitch` the export stays on the old publisher while it is +up. Only lite-07 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`. ## Debugging diff --git a/doc/bin/srt.md b/doc/bin/srt.md index fd5688c692..1a07d87303 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -27,7 +27,10 @@ moq --connect https://relay.example.com/anon --broadcast event.hang import srt - 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. +broadcast restart, and their next subscribe reaches the new feed. An +`export srt` stream ends with its broadcast; `--linger` and `--stitch` follow a +return or a replacement on the same connection, as they do for +[`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/upgrade.md b/doc/setup/upgrade.md index 2f62605464..f1088eeec8 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -30,6 +30,12 @@ These land with the next breaking release, not the 2026-09-23 train. moq-lite 06 and older and moq-transport. moq-lite 07 adds `ANNOUNCE_RESTART` (0x3), and lite-06's `ANNOUNCE_RESTART` is named `ANNOUNCE_UPDATE`, as it always meant. +- **`moq export ts --linger` waits only for the same publisher instance.** A + restarted publisher (a new epoch, or any return on a version without one) + exits 1 unless `--stitch` follows it as a program switch. `export srt` takes + the same two flags. In Rust, moq-mux's `ts::Export::resume` is + `ts::Export::follow`, `Source::returned` is gone, and `ts::Follower` follows a + path's announcements the way both exports do. - **fMP4 export of Annex-B H.264 and H.265 inits from the catalog.** When the catalog codec string and dimensions are enough, `moq export fmp4` writes an `avc3` or `hev1` init segment before the first keyframe and leaves SPS, PPS, diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 11936673ad..0c6af98641 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -107,7 +107,6 @@ maintainer's pre-release merge of `release` into `main` keeps `main`'s - [JS consume identity](/quest/m0/broadcast-epoch/js-consume-identity.md) - requests carry the serving prefix's epoch and new winners cannot reuse an old cached broadcast - [JS restart keeps the request](/quest/m0/broadcast-epoch/js-restart-keeps-request.md) - a resolved request remains on its old instance, matching Rust's sticky subscriptions -- [Export ts](/quest/m0/broadcast-epoch/export-ts.md) - `export ts` and SRT egress linger only for the same epoch, and switch to a replacement only with `--stitch`, as a full program switch - [Publish catalog restart](/quest/m0/broadcast-epoch/publish-catalog-restart.md) - `@moq/publish` never reuses catalog group numbers under one name and epoch after a re-announce - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and let a publisher announce one - [moqsrc](/quest/m0/broadcast-epoch/moqsrc.md) - moqsrc switches to the new broadcast on a `Restart`, keeping its pads by rendition name diff --git a/quest/m0/broadcast-epoch/export-ts.md b/quest/m0/broadcast-epoch/export-ts.md deleted file mode 100644 index bcb34880d7..0000000000 --- a/quest/m0/broadcast-epoch/export-ts.md +++ /dev/null @@ -1,105 +0,0 @@ -# [M] TS export lingers within an epoch and stitches across one only on request - -## Goal - -`moq export ts` and moq-srt's egress never splice two broadcasts into one TS -stream unless asked: - -- `--linger ` waits up to that long for the *same* broadcast (the same - epoch) to come back after its route goes, and carries on with the same - stream. -- A replacement (a newer epoch, or another instance of an epochless route) - ends the export with an error that names `--stitch`. On an epochless route - every return is a replacement. -- `--stitch` opts in to following a replacement. The export is rebuilt on the - new broadcast with a full program switch: a new PMT from the new catalog, so tracks, codecs, and PIDs may - change, plus PCR and continuity discontinuities. Nothing from the old - broadcast survives except the output pipe or SRT connection, and the - PSI version numbers. - -Non-goal: identical output across a replacement. Two legs that are -byte-identical within an epoch may differ after one (#5101). - -## Plan - -Decided 2026-10-09. Today `--linger` (#4504) waits for the ended broadcast to -close and the path to route again, then calls `ts::Export::resume()`. That -call clears the decode clocks, jitter buffer, and schedule, but keeps the old -PMT and PIDs and matches the returned tracks by name only, so a renamed codec -goes out under the old stream_type and continuity counters run on unflagged. -That is a splice of whatever returns, assuming the layout matches without -checking. It also keys on errors and the broadcast closing, so an old -publisher that stays alive holds the export on a replaced broadcast. - -- Delete `Export::resume()` and its partial reset. -- Drive both flags from announcements, as players do: `Start`, `Restart`, and - `End` with their epochs. An export that fails while its broadcast is still - announced exits 1 without lingering, as today. -- Use `origin.follow(path)` for the resolved path events, shared by CLI and - SRT. The 2026-10-10 audit of #5147 found a second prefix watcher treating - any covering prefix's End or Restart as a change even while an exact - route still wins. Regress with a live exact route while a broader prefix - starts, restarts, or ends: none changes the export. Actual winner changes - and gaps must still drive the policy above. Coordinate with the existing - follower-gap fix instead of reimplementing route resolution here. -- Same epoch within `--linger`: `follow` continues the same stream. - It is the same content, so the existing skip handling covers the gap (late - frames drop, and a forward leap opens a new generation). Within an epoch, - moq-net already resumes across routes, so this only matters once every - route has gone. -- Without `--stitch`: subscriptions are sticky, so a `Restart` leaves the - export on the old broadcast until it ends. If the path then serves a - different instance, the export exits 1 with an error naming `--stitch`. -- With `--stitch`: on `Restart` (or `End` then `Start` with another - instance) `follow` it at once, which builds a fresh `Export`. Every PID's first packet carries - `discontinuity_indicator`, the PAT and PMT `version_number`s advance (the only - values handed from the old `Export`, since a version-caching demux such as - TSDuck's ignores a changed PAT under an unchanged version), and each - elementary stream starts at a keyframe. Rejected: continuing only when the layout is identical, and - deleting `--linger`. Exiting does not hide the switch from a receiver - downstream of `tsp`, it only drops the flags. -- moq-srt's egress (`rs/moq-srt/src/ts.rs`) gets the same two options and keeps - its SRT connection across a stitch. Its linger defaults to 0, like the CLI: - an `End` that nothing replaces within the linger closes the SRT stream, so - a finished broadcast never leaves a caller connected indefinitely. -- Decided: the SRT policy lives on `moq_srt::Config` as `linger: Duration` - (default 0) and `stitch: bool` (default false), mirrored as CLI flags. -- Decided: one successor operation, `ts::Export::follow(self, broadcast) -> - Export`, used by both callers. On the same epoch it continues the stream - with the same PSI. On another instance it does the full program switch, - carrying the PAT and PMT `version_number`s, both advanced. The caller calls - it for a replacement only under `--stitch`. Rejected: separate continue and - stitch methods, and a builder policy that has the export watch announces - itself. -- Decided: the TS output has no epoch. A re-import of an export is out of - scope. -- Fold in the stale wording from the marker audit: a marker group only - declares a pause or forward break, since rewinds are refused (#3711). Rename - `Consumer::restarts` / `ExportSource::restarts` for what they count, and fix - the comments that say the publisher restarted or rewound its timeline - (`rs/moq-mux/src/container/consumer.rs`, `rs/moq-srt/src/server.rs`). -- Tests, with mocked time: - - A same-epoch return within the linger continues the stream with no PSI - version change. - - A replacement without `--stitch` exits 1, and an old publisher that stays - alive keeps the export on it until it ends. - - With `--stitch`, a replacement with a different codec and track set - writes a new PMT version, flags every PID's first packet, starts each - stream at a keyframe, and writes nothing from the old broadcast after the - break. - - A stitch that keeps the transport-stream ID but moves the PMT PID - advances the PAT version. - - Update #4504's relay-backed CLI tests, and add the stitch case for SRT - egress, plus an unreplaced `End` closing the SRT stream at the linger. -- Docs: the `--linger` paragraph in `doc/bin/cli.md` (plus `--stitch`), any - restart behavior in `doc/bin/srt.md`, and every example invocation using - `--linger`. - -Public API: breaking, `ts::Export::resume` is removed and `restarts()` is -renamed. New: `ts::Export::follow`, and `linger` and `stitch` on -`moq_srt::Config`. CLI: new `--stitch` on `export ts` and SRT egress, and `--linger` -no longer follows a replacement. Wire: none. - -## Related - -- [Two TS export legs](/quest/m2/ts-hitless.md) - identical output within one epoch diff --git a/quest/m2/ts-hitless.md b/quest/m2/ts-hitless.md index b991c13f8d..4054091890 100644 --- a/quest/m2/ts-hitless.md +++ b/quest/m2/ts-hitless.md @@ -8,7 +8,7 @@ skips and source drift, continuity counters included, within one epoch. A ST 202 can then switch between them hitlessly. Non-goal: identity across a replaced broadcast. A leg ends on a replacement, or -starts a fresh stream with `--stitch` ([Export ts](/quest/m0/broadcast-epoch/export-ts.md)), +starts a fresh stream with `--stitch`, so legs may differ after one (#5101, maintainer 2026-10-09: byte-identical output across restarts is not supported). diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 2831a60755..25ae9ccfbd 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -937,7 +937,7 @@ pub enum ExportSink { /// RTMP: push to a remote (`--connect`) or serve plays (`--listen`). Rtmp(crate::rtmp::ExportArgs), /// SRT: push to a remote (`--connect`) or serve requests (`--listen`). - Srt(crate::srt::Args), + Srt(crate::srt::ExportArgs), /// WebRTC: WHIP client pushing to a remote (`--connect`) or WHEP server serving plays (`--listen`). Rtc(crate::rtc::Args), /// Record the broadcast into an object store until it ends. @@ -970,6 +970,7 @@ impl ExportSink { format, max_delay: container.max_delay.into_std(), linger: container.linger.into_std(), + stitch: false, fragment_duration: None, mux_rate: None, }; @@ -986,6 +987,7 @@ impl ExportSink { format: SubscribeFormat::Ts, max_delay: args.delay.into_std(), linger: args.linger.into_std(), + stitch: args.stitch, fragment_duration: None, mux_rate: args.mux_rate, }, @@ -1003,6 +1005,8 @@ pub struct Stdout { /// The staleness budget, which `ts` also holds every frame for (`--delay`). pub max_delay: Duration, pub linger: Duration, + /// Follow a replacement as a program switch (`ts` only). + pub stitch: bool, pub fragment_duration: Option, pub mux_rate: Option, } @@ -1052,12 +1056,17 @@ pub struct Transport { #[usage(long, default = "500ms")] pub delay: crate::duration::Duration, - /// How long to wait for the broadcast to come back once it ends (e.g. `10s`). - /// The output stops while it is gone and resumes flagged as a break. + /// 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. #[usage(long, default = "0s")] pub linger: crate::duration::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. + #[usage(long)] + pub stitch: bool, + /// Pad the output with null packets to this constant rate, in bits per second. /// Defaults to the multiplex rate the catalog recorded from a constant-rate /// source (`mpegts.muxRate`); without either the output is unpadded. @@ -1328,6 +1337,34 @@ mod tests { ); } + /// `export srt` follows its broadcast with the same `--linger` and `--stitch` as `export ts`. + #[test] + fn export_srt_takes_linger_and_stitch() { + let cli = Invocation::try_parse_from([ + "moq", + "export", + "srt", + "--listen", + "[::]:9000", + "--linger", + "10s", + "--stitch", + ]) + .unwrap(); + let Command::Export(export) = &cli.stages[0] else { + panic!("an export stage"); + }; + let ExportSink::Srt(args) = &export.sink else { + panic!("an export srt stage"); + }; + assert_eq!(args.linger.into_std(), Duration::from_secs(10)); + assert!(args.stitch); + assert!( + Invocation::try_parse_from(["moq", "import", "srt", "--listen", "[::]:9000", "--stitch"]).is_err(), + "an ingest has nothing to follow" + ); + } + /// A released spelling is refused, and the error names what to write instead. /// /// The alternative is what this replaced: the flag parsed onto a hidden field, diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 03c45e6e15..aa3f29d199 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -742,6 +742,7 @@ fn spawn_export( format: stdout.format, max_delay: stdout.max_delay, linger: stdout.linger, + stitch: stdout.stitch, fragment_duration: stdout.fragment_duration, mux_rate: stdout.mux_rate, catalog: export.catalog_format, @@ -765,11 +766,11 @@ fn spawn_export( } } ExportSink::Srt(srt) => { - if let Some(addr) = srt.listen { + if let Some(addr) = srt.endpoint.listen { let name = require_broadcast(name, "export srt --listen")?; - tasks.spawn(srt::listen_export(origin.consume(), addr, name, srt.latency.into_std())); - } else if let Some(url) = srt.connect { - tasks.spawn(srt::connect_export(origin.consume(), url, name, srt.latency.into_std())); + tasks.spawn(srt::listen_export(origin.consume(), addr, name, srt)); + } else if let Some(url) = srt.endpoint.connect.clone() { + tasks.spawn(srt::connect_export(origin.consume(), url, name, srt)); } } ExportSink::Rtc(rtc) => { @@ -816,8 +817,7 @@ async fn run_stdout(consumer: moq_net::origin::Consumer, name: String, args: Sub .await .ok_or_else(|| anyhow::anyhow!("origin closed before broadcast `{name}` was announced"))?; - let source = moq_mux::Source::new(consumer, &name); - Subscribe::new(source, catalog, args).run().await + Subscribe::new(consumer, &name, catalog, args).run().await } /// Run every endpoint until the first finishes (stdin EOF, SIGINT, SIGTERM, or an diff --git a/rs/moq-cli/src/srt.rs b/rs/moq-cli/src/srt.rs index 7935de7553..582fd3a588 100644 --- a/rs/moq-cli/src/srt.rs +++ b/rs/moq-cli/src/srt.rs @@ -46,6 +46,25 @@ pub struct ImportArgs { pub program: Option, } +/// SRT export args: the endpoint, plus how a broadcast ending or being replaced is followed. +#[derive(usage::Args, Clone)] +#[usage(unknown_flags = "error", args_override_self = false)] +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. + #[usage(long, default = "0s")] + pub linger: crate::duration::Duration, + + /// Follow another publisher instance that replaces the broadcast, as a full program switch + /// on the same SRT connection. + #[usage(long)] + pub stitch: bool, +} + impl ImportArgs { /// The library's selection for `--program`. pub fn program(&self) -> Option { @@ -121,9 +140,10 @@ pub async fn listen_export( origin: moq_net::origin::Consumer, addr: SocketAddr, name: String, - latency: Duration, + args: ExportArgs, ) -> anyhow::Result<()> { - let mut server = Server::bind(addr, latency).await?; + let mut server = Server::bind(addr, args.endpoint.latency.into_std()).await?; + let (linger, stitch) = (args.linger.into_std(), args.stitch); tracing::info!(%addr, %name, "SRT listening (export)"); notify_ready(); @@ -133,6 +153,7 @@ pub async fn listen_export( let origin = origin.clone(); let name = name.clone(); tokio::spawn(async move { + let subscribe = subscribe.with_linger(linger).with_stitch(stitch); if let Err(err) = subscribe.accept(&origin, &name).await { tracing::warn!(%name, %err, "SRT request ended with error"); } @@ -175,13 +196,16 @@ pub async fn connect_export( origin: moq_net::origin::Consumer, url: Url, name: String, - latency: Duration, + args: ExportArgs, ) -> anyhow::Result<()> { let (addr, resource) = parse_url(&url).await?; tracing::info!(url = %RedactedUrl::new(&url), %name, "SRT client pushing"); notify_ready(); - let client = moq_srt::Client::new(addr, resource).with_latency(latency); + let client = moq_srt::Client::new(addr, resource) + .with_latency(args.endpoint.latency.into_std()) + .with_linger(args.linger.into_std()) + .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 7aac36eed8..05cb099711 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -1,6 +1,8 @@ use std::time::Duration; +use anyhow::Context; use hang::catalog::{AudioCodecKind, VideoCodecKind}; +use hang::moq_net; use moq_mux::catalog::{self, CatalogFormat, Stream}; use moq_mux::select; use tokio::io::AsyncWriteExt; @@ -182,6 +184,9 @@ pub struct SubscribeArgs { /// How long to wait for the broadcast to come back after it ends (TS only). pub linger: Duration, + /// Follow another publisher instance replacing the broadcast, as a program switch (TS only). + pub stitch: bool, + /// Cap the output duration: publisher groups by default for fMP4, video GOPs for MKV. pub fragment_duration: Option, @@ -240,15 +245,26 @@ impl SubscribeArgs { /// Exports one broadcast from the Origin to stdout in the requested format. pub struct Subscribe { + origin: moq_net::origin::Consumer, + path: moq_net::PathOwned, source: moq_mux::Source, catalog: CatalogFormat, args: SubscribeArgs, } impl Subscribe { - /// Wrap the broadcast + resolved settings; [`run`](Self::run) drives it. - pub fn new(source: moq_mux::Source, catalog: CatalogFormat, args: SubscribeArgs) -> Self { - Self { source, catalog, args } + /// Export the broadcast at `path` on `origin` with the resolved settings; [`run`](Self::run) + /// drives it. + pub fn new(origin: moq_net::origin::Consumer, path: &str, catalog: CatalogFormat, args: SubscribeArgs) -> Self { + let path = moq_net::Path::new(path).to_owned(); + let source = moq_mux::Source::new(origin.clone(), &path); + Self { + origin, + path, + source, + catalog, + args, + } } /// Build the catalog stream, narrowed by the rendition selection flags. The @@ -338,13 +354,18 @@ impl Subscribe { async fn run_ts(self) -> anyhow::Result<()> { let mut stdout = tokio::io::stdout(); + self.origin.routed(&self.path).await.with_context(|| { + format!( + "broadcast `{}` is outside the session's scope, or the origin closed before it was announced", + self.path + ) + })?; + // TS emits PAT/PMT then a continuous PES stream (re-emitting PAT/PMT at // keyframes for tune-in). Avc3/Hev1 sources pass through as Annex-B; AAC // is re-framed as ADTS. `fragment_duration` does not apply to TS. `with_ts` // selects the `mpegts` catalog extension so undecoded elementary streams // (SCTE-35, teletext, DVB AC-3, ...) are re-emitted verbatim on their PIDs. - let source = self.source.clone(); - let mut broadcast = source.broadcast().await?; let mut ts = moq_mux::container::ts::Export::with_ts(self.source, self.catalog) .await? .with_delay(self.args.max_delay); @@ -355,8 +376,11 @@ impl Subscribe { // A TS byte stream carries no per-frame timing, so delivery time is the only // carrier of each frame's spacing (#2984). The export lays each slice of the PCR // grid out at its time on its own clock, which follows the source's, so each is - // written as it comes. - let linger = self.args.linger; + // 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_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. let mut log = moq_mux::container::ts::stats::Log::default(); @@ -371,52 +395,31 @@ impl Subscribe { "TS export release clock" ); }; - loop { - let end = loop { - let frame = match ts.next().await { - Ok(Some(frame)) => frame, - Ok(None) => break Ok(()), - Err(err) => break Err(err), - }; - stdout.write_all(&frame.payload).await?; - stdout.flush().await?; - - if sampled.elapsed() >= moq_mux::container::ts::stats::Log::INTERVAL { - sampled = tokio::time::Instant::now(); - let stats = ts.stats(); - if (stats.dropped, stats.out_of_tolerance) != reported { - reported = (stats.dropped, stats.out_of_tolerance); - release(&stats); - } - log.sample(stats.into()); - } + let end = loop { + let frame = match ts.next().await { + Ok(Some(frame)) => frame, + Ok(None) => break Ok(()), + Err(err) => break Err(err), }; + stdout.write_all(&frame.payload).await?; + stdout.flush().await?; - release(&ts.stats()); - - // An end waits out the linger, and on expiry the last one is the result: a - // clean catalog finish exits 0, a drop or any other failure exits 1. - if linger.is_zero() { - return Ok(end?); - } - // A failure ends the broadcast only if the broadcast goes too. One that stays - // up cannot return, so the failure is the export's own and exits now. - if let Err(err) = &end - && !closes_within(&broadcast, CLOSE_GRACE.min(linger)).await - { - tracing::warn!(%err, "export failed with the broadcast still up, so not lingering"); - return Ok(end?); + if sampled.elapsed() >= moq_mux::container::ts::stats::Log::INTERVAL { + sampled = tokio::time::Instant::now(); + let stats = ts.export().stats(); + if (stats.dropped, stats.out_of_tolerance) != reported { + reported = (stats.dropped, stats.out_of_tolerance); + release(&stats); + } + log.sample(stats.into()); } - match &end { - Ok(()) => tracing::info!(?linger, "broadcast finished, waiting for it to return"), - Err(err) => tracing::warn!(%err, ?linger, "broadcast ended, waiting for it to return"), + }; + release(&ts.export().stats()); + match end { + Err(err @ moq_mux::Error::Replaced(_)) => { + Err(anyhow::Error::from(err).context("pass --stitch to follow a replacement as a program switch")) } - let Some(returned) = resume_within(&source, &broadcast, &mut ts, linger).await? else { - tracing::info!(?linger, "broadcast did not return"); - return Ok(end?); - }; - broadcast = returned; - tracing::info!("broadcast returned, resuming"); + end => Ok(end?), } } @@ -438,99 +441,3 @@ impl Subscribe { Ok(()) } } - -/// How long an export failure waits for its broadcast to close before it counts as the export's own. -/// -/// A killed publisher's tracks can error just before its broadcast closes, so the two need not -/// land together. -const CLOSE_GRACE: Duration = Duration::from_secs(1); - -/// Whether `broadcast` closes within `grace`. -async fn closes_within(broadcast: &hang::moq_net::broadcast::Consumer, grace: Duration) -> bool { - tokio::time::timeout(grace, broadcast.closed()).await.is_ok() -} - -/// Wait up to `linger` for the `ended` broadcast to return and `ts` to resume on it. -/// -/// The linger bounds the whole return, catalog subscription included: a returned -/// broadcast whose catalog never resolves must not hold the export past it. `None` -/// when it did not return in time. -async fn resume_within( - source: &moq_mux::Source, - ended: &hang::moq_net::broadcast::Consumer, - ts: &mut moq_mux::container::ts::Export, - linger: Duration, -) -> anyhow::Result> { - let resume = async { - let returned = source.returned(ended).await?; - ts.resume().await?; - anyhow::Ok(returned) - }; - match tokio::time::timeout(linger, resume).await { - Ok(returned) => Ok(Some(returned?)), - Err(_) => Ok(None), - } -} - -#[cfg(test)] -mod tests { - use super::*; - - /// A broadcast that stays up is not closing, however long the grace. - #[tokio::test(start_paused = true)] - async fn a_live_broadcast_does_not_close_within_the_grace() { - let (origin, driver) = hang::moq_net::origin::Producer::new(Default::default()); - tokio::spawn(hang::moq_net::time::run(driver)); - let _live = origin.publish("live", Default::default()).unwrap(); - let broadcast = origin.consume().request_broadcast("live", None).await.unwrap(); - - let start = tokio::time::Instant::now(); - assert!(!closes_within(&broadcast, CLOSE_GRACE).await); - assert_eq!(start.elapsed(), CLOSE_GRACE); - } - - /// A broadcast that closes just after its tracks fail is still an end. - #[tokio::test(start_paused = true)] - async fn a_close_just_after_the_failure_is_an_end() { - let (origin, driver) = hang::moq_net::origin::Producer::new(Default::default()); - tokio::spawn(hang::moq_net::time::run(driver)); - let live = origin.publish("live", Default::default()).unwrap(); - let broadcast = origin.consume().request_broadcast("live", None).await.unwrap(); - - let gap = CLOSE_GRACE / 4; - tokio::spawn(async move { - tokio::time::sleep(gap).await; - drop(live); - }); - let start = tokio::time::Instant::now(); - assert!(closes_within(&broadcast, CLOSE_GRACE).await); - assert_eq!(start.elapsed(), gap); - } - - /// A broadcast that returns but never serves its catalog gives up at the linger, - /// rather than waiting on the catalog past it. - #[tokio::test(start_paused = true)] - async fn a_return_without_a_catalog_expires_with_the_linger() { - let (origin, driver) = hang::moq_net::origin::Producer::new(Default::default()); - tokio::spawn(hang::moq_net::time::run(driver)); - let source = moq_mux::Source::new(origin.consume(), "live"); - - let mut first = origin.publish("live", Default::default()).unwrap(); - let catalog = moq_mux::catalog::Producer::new(&mut first, Default::default()).unwrap(); - let ended = source.broadcast().await.unwrap(); - let mut ts = moq_mux::container::ts::Export::with_ts(source.clone(), CatalogFormat::Hang) - .await - .unwrap(); - drop((first, catalog)); - - // Back, but its catalog request is never answered. - let second = origin.publish("live", Default::default()).unwrap(); - let _unanswered = second.dynamic(); - - let linger = Duration::from_secs(10); - let start = tokio::time::Instant::now(); - let resumed = resume_within(&source, &ended, &mut ts, linger).await.unwrap(); - assert!(resumed.is_none(), "a return that never resumes is no return"); - assert_eq!(start.elapsed(), linger); - } -} diff --git a/rs/moq-cli/tests/export.rs b/rs/moq-cli/tests/export.rs index d0c703ea67..3108c9a1ad 100644 --- a/rs/moq-cli/tests/export.rs +++ b/rs/moq-cli/tests/export.rs @@ -1,5 +1,6 @@ -//! `moq export ts --linger` over a real relay: the export rides out a publisher that -//! leaves and comes back, and exits with the verdict of the broadcast's last end. +//! `moq export ts --linger` 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. #![cfg(unix)] use std::process::Stdio; @@ -20,21 +21,31 @@ type Output = Arc>>; struct Relay { url: String, - fingerprint: String, } +/// A public relay on ephemeral loopback ports that also accepts lite-07, the version that +/// carries epochs, for a client that opts in with `--connect-version`. Its generated +/// certificate is trusted blindly, since nothing but loopback reaches it. async fn relay() -> Relay { let _ = moq_tokio::crypto::install_default(); - let fixture = moq_relay::test_relay().await.expect("test relay"); - let ready = fixture.relay.ready(); - tokio::spawn(fixture.relay.run()); + let mut config = moq_relay::Config::default(); + config.drain_timeout = Duration::ZERO; + config.listen.bind = Some("127.0.0.1:0".parse().unwrap()); + config.listen.tls.generate = vec!["localhost".into()]; + config.listen.version = std::iter::once(LITE_07.parse().unwrap()) + .chain(hang::moq_net::Versions::all().iter().copied()) + .collect(); + config.auth.public = vec![moq_auth::Pattern::all()]; + let relay = moq_relay::Relay::load(config).await.expect("test relay"); + let url = format!("https://{}/", relay.quic_addr().expect("QUIC address")); + let ready = relay.ready(); + tokio::spawn(relay.run()); ready.wait().await.expect("relay ready"); - Relay { - url: fixture.url.to_string(), - fingerprint: fixture.fingerprint, - } + Relay { url } } +const LITE_07: &str = "moq-lite-07-wip"; + fn moq(relay: &Relay, args: &[&str]) -> tokio::process::Command { let mut command = tokio::process::Command::new(env!("CARGO_BIN_EXE_moq")); // A `MOQ_*` variable in the developer's shell must not reconfigure the child. @@ -44,14 +55,7 @@ fn moq(relay: &Relay, args: &[&str]) -> tokio::process::Command { } } command - .args([ - "--connect", - &relay.url, - "--connect-tls-fingerprint", - &relay.fingerprint, - "--broadcast", - "demo", - ]) + .args(["--connect", &relay.url, "--connect-tls-insecure", "--broadcast", "demo"]) .args(args) .kill_on_drop(true); command @@ -59,28 +63,55 @@ fn moq(relay: &Relay, args: &[&str]) -> tokio::process::Command { /// Start `moq export ts --linger `, collecting its stdout. fn export(relay: &Relay, linger: &str) -> (Child, Output) { - let mut child = moq(relay, &["export", "ts", "--linger", linger]) + let (child, output, _) = export_with(relay, &[], &["--linger", linger]); + (child, output) +} + +/// Start `moq export ts `, collecting its stdout, and its stderr until it exits. +fn export_with(relay: &Relay, global: &[&str], flags: &[&str]) -> (Child, Output, JoinHandle) { + let mut child = moq(relay, global) + .args(["export", "ts"]) + .args(flags) .stdout(Stdio::piped()) + .stderr(Stdio::piped()) .spawn() .expect("spawn export"); - let mut stdout = child.stdout.take().expect("stdout"); + let output = collect(child.stdout.take().expect("stdout")); + // Read to the end, since the exit error is the last thing written before the pipe closes. + let mut stderr = child.stderr.take().expect("stderr"); + let errors = tokio::spawn(async move { + let mut errors = Vec::new(); + let _ = stderr.read_to_end(&mut errors).await; + String::from_utf8_lossy(&errors).into_owned() + }); + (child, output, errors) +} + +/// Collect everything `pipe` yields. +fn collect(mut pipe: impl tokio::io::AsyncRead + Unpin + Send + 'static) -> Output { let output = Output::default(); let sink = output.clone(); tokio::spawn(async move { let mut buf = vec![0; 64 * 1024]; - while let Ok(n) = stdout.read(&mut buf).await + while let Ok(n) = pipe.read(&mut buf).await && n > 0 { sink.lock().unwrap().extend_from_slice(&buf[..n]); } }); - (child, output) + output } /// Start `moq import ts` and feed it the clip at about real time, handing stdin back /// once it is all written. Closing stdin then finishes the broadcast. fn import(relay: &Relay) -> (Child, JoinHandle) { - let mut child = moq(relay, &["import", "ts"]) + import_with(relay, &[]) +} + +/// [`import`] with `global` flags, such as the session version or the epoch. +fn import_with(relay: &Relay, global: &[&str]) -> (Child, JoinHandle) { + let mut child = moq(relay, global) + .args(["import", "ts"]) .stdin(Stdio::piped()) .spawn() .expect("spawn import"); @@ -207,18 +238,73 @@ async fn an_export_failure_with_the_broadcast_up_exits_without_lingering() { wait(&mut publisher).await; } -#[tokio::test] -async fn a_publisher_restarted_within_the_linger_resumes_the_output() { - let relay = relay().await; - let (mut export, output) = export(&relay, "10s"); +/// The `version_number` of a packet starting a PAT section. +fn pat_version(packet: &[u8]) -> Option { + if pid(packet) != 0 || packet[1] & 0x40 == 0 { + return None; + } + let start = 4 + adaptation_flags(packet).map_or(0, |_| 1 + usize::from(packet[4])); + let section = start + 1 + usize::from(packet[start]); + Some((packet[section + 5] >> 1) & 0x1f) +} - let (publisher, feeding) = import(&relay); +/// Each PID whose first packet in `ts` does not flag a discontinuity. +fn unflagged(ts: &[u8]) -> Vec { + let mut seen = std::collections::HashSet::new(); + packets(ts) + .filter(|packet| pid(*packet) != 0x1fff && seen.insert(pid(*packet))) + .filter(|packet| adaptation_flags(*packet).is_none_or(|flags| flags & 0x80 == 0)) + .map(|packet| pid(packet)) + .collect() +} + +/// Run `publisher` until it was interrupted, then mark the end of its output once the paced +/// tail has drained. +async fn interrupted(output: &Output, publisher: (Child, JoinHandle)) -> usize { + let (publisher, feeding) = publisher; let stdin = feeding.await.unwrap(); - output_past(&output, 0).await; + output_past(output, 0).await; interrupt(publisher, stdin).await; - // Let the paced tail of the first broadcast drain before marking where it ended. tokio::time::sleep(Duration::from_secs(1)).await; - let mark = output.lock().unwrap().len().next_multiple_of(188); + output.lock().unwrap().len().next_multiple_of(188) +} + +/// A restarted publisher is another instance, so the export does not splice it in: it exits 1 +/// as soon as the replacement shows up, naming the flag that would follow it. +#[tokio::test] +async fn a_restarted_publisher_exits_one_without_stitch() { + let relay = relay().await; + let (mut export, output, errors) = export_with(&relay, &[], &["--linger", "20s"]); + let mark = interrupted(&output, import(&relay)).await; + + let (mut publisher, feeding) = import(&relay); + let started = Instant::now(); + let status = wait(&mut export).await; + assert_eq!(status.code(), Some(1), "a replacement exits 1, got {status}"); + assert!( + started.elapsed() < Duration::from_secs(10), + "exited after {:?}, as though waiting out the linger", + started.elapsed() + ); + let errors = errors.await.unwrap(); + assert!(errors.contains("--stitch"), "the error names --stitch: {errors}"); + assert_eq!( + output.lock().unwrap().len(), + mark, + "nothing of the replacement went out" + ); + + drop(feeding.await.unwrap()); + wait(&mut publisher).await; +} + +/// With `--stitch`, a restarted publisher is a full program switch: a new PAT version, and +/// every PID's first packet flagging the break. +#[tokio::test] +async fn a_restarted_publisher_is_a_program_switch_with_stitch() { + let relay = relay().await; + let (mut export, output, _) = export_with(&relay, &[], &["--linger", "10s", "--stitch"]); + let mark = interrupted(&output, import(&relay)).await; let (mut publisher, feeding) = import(&relay); let stdin = feeding.await.unwrap(); @@ -230,13 +316,95 @@ async fn a_publisher_restarted_within_the_linger_resumes_the_output() { assert!(status.success(), "the last end was a clean finish, got {status}"); let output = output.lock().unwrap(); - let after = &output[mark..]; + let (before, after) = output.split_at(mark); let total = packets(after).count(); - assert!(total > 100, "the returned broadcast went out: {total} packets"); - assert!(packets(after).any(|p| pid(p) == 0), "PAT re-emitted after the restart"); - // The PCR the returned broadcast opens with flags the break. - let first_pcr = packets(after) - .find_map(|p| adaptation_flags(p).filter(|flags| flags & 0x10 != 0)) - .expect("a PCR after the restart"); - assert_ne!(first_pcr & 0x80, 0, "the restart is flagged as a break"); + assert!(total > 100, "the replacement went out: {total} packets"); + assert!( + packets(before) + .filter_map(|p| pat_version(p)) + .all(|version| version == 0) + ); + let versions: Vec = packets(after).filter_map(|p| pat_version(p)).collect(); + assert!( + !versions.is_empty() && versions.iter().all(|&version| version == 1), + "the switch advances the PAT version: {versions:?}" + ); + assert_eq!(unflagged(after), Vec::::new(), "every PID flags the break"); +} + +/// The same instance (a shared `--epoch` on a version that carries it) back within the linger +/// continues the stream, under the program already announced. +#[tokio::test] +async fn the_same_instance_within_the_linger_continues() { + let relay = relay().await; + let epoch = hang::moq_net::Epoch::mint().to_string(); + let lite07 = ["--connect-version", LITE_07]; + let publish = [lite07[0], lite07[1], "--epoch", epoch.as_str()]; + let (mut export, output, _) = export_with(&relay, &lite07, &["--linger", "10s"]); + let mark = interrupted(&output, import_with(&relay, &publish)).await; + + let (mut publisher, feeding) = import_with(&relay, &publish); + let stdin = feeding.await.unwrap(); + output_past(&output, mark).await; + drop(stdin); + assert!(wait(&mut publisher).await.success()); + + let status = wait(&mut export).await; + assert!(status.success(), "the last end was a clean finish, got {status}"); + + let output = output.lock().unwrap(); + let after = &output[mark..]; + assert!(packets(after).count() > 100, "the returned broadcast went out"); + let versions: Vec = packets(&output).filter_map(|p| pat_version(p)).collect(); + assert!( + versions.iter().all(|&version| version == 0), + "the same instance keeps the PSI version: {versions:?}" + ); +} + +/// Subscriptions are sticky: a replacement announced while the old publisher stays up leaves +/// the export on the old one until it ends, and only then exits 1. +#[tokio::test] +async fn an_old_publisher_that_stays_up_keeps_the_export() { + let relay = relay().await; + let lite07 = ["--connect-version", LITE_07]; + let (mut export, output, errors) = export_with(&relay, &lite07, &["--linger", "10s"]); + + let (mut old, feeding) = import_with(&relay, &lite07); + output_past(&output, 0).await; + let (mut new, replacing) = import_with(&relay, &lite07); + let old_stdin = feeding.await.unwrap(); + let new_stdin = replacing.await.unwrap(); + tokio::time::sleep(Duration::from_secs(1)).await; + assert!( + export.try_wait().unwrap().is_none(), + "the export stays on the old publisher while it is up" + ); + let written = output.lock().unwrap().len(); + assert!( + packets(&output.lock().unwrap()) + .filter_map(|p| pat_version(p)) + .all(|v| v == 0) + ); + + drop(old_stdin); + assert!(wait(&mut old).await.success()); + let status = wait(&mut export).await; + assert_eq!( + status.code(), + Some(1), + "the replacement exits 1 once the old one ends, got {status}" + ); + let errors = errors.await.unwrap(); + assert!(errors.contains("--stitch"), "the error names --stitch: {errors}"); + let output = output.lock().unwrap().clone(); + assert!( + packets(&output[written.next_multiple_of(188).min(output.len())..]) + .filter_map(|p| pat_version(p)) + .all(|version| version == 0), + "nothing of the replacement went out" + ); + + drop(new_stdin); + wait(&mut new).await; } diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index 7fc0e95eb0..bf6f4fe2d7 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -95,8 +95,8 @@ pub struct Consumer { // Increments on a declared marker, an unproven delivered hole, and a latency skip. discontinuity: u64, - // Increments on a declared marker only: the publisher restarted its timeline. - restarts: u64, + // Increments on a declared marker only: the publisher paused or broke its timeline forward. + markers: u64, // Exclusive audio endpoint delivered before terminal codec packets. end: Option, @@ -143,7 +143,7 @@ impl Consumer { floor: None, presented_end: None, discontinuity: 0, - restarts: 0, + markers: 0, end: None, } } @@ -157,11 +157,12 @@ impl Consumer { self.discontinuity } - /// A counter that increments only when the publisher declared a marker group: its - /// timeline restarted. A hole or a latency skip moves the playhead - /// ([`Self::discontinuity`]) but the timeline carries on, so this stays put. - pub(crate) fn restarts(&self) -> u64 { - self.restarts + /// A counter that increments only when the publisher declared a marker group: a pause or + /// a forward break in its timeline, since a rewind is refused. A hole or a latency skip + /// moves the playhead ([`Self::discontinuity`]) but the timeline carries on, so this + /// stays put. + pub(crate) fn markers(&self) -> u64 { + self.markers } /// The exclusive audio endpoint delivered before terminal codec packets. @@ -313,7 +314,7 @@ impl Consumer { self.note_group_edge(); if marker { self.bump_playhead(); - self.restarts += 1; + self.markers += 1; } return Poll::Ready(Ok(Some(Event::GroupEnd))); } @@ -403,7 +404,7 @@ impl Consumer { if hole || had_marker { self.bump_playhead(); } - self.restarts += u64::from(had_marker); + self.markers += u64::from(had_marker); let new_current = self.pending.front().map(|g| g.sequence).unwrap(); tracing::debug!(old = self.current, new = new_current, "skipping slow groups"); @@ -920,10 +921,10 @@ mod tests { assert_eq!(consumer.discontinuity(), 1); } - /// A missing group moves the playhead, but the timeline carries on: it is not a restart. + /// A missing group moves the playhead, but the timeline carries on: it is not a marker. /// A declared marker is both. #[tokio::test] - async fn only_a_marker_restarts_the_timeline() { + async fn only_a_marker_breaks_the_timeline() { let mut track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.audio)); let consumer_track = track.subscribe(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(2))); @@ -939,10 +940,10 @@ mod tests { assert_eq!(consumer.read().await.unwrap().unwrap().timestamp, ts(0)); assert_eq!(consumer.read().await.unwrap().unwrap().timestamp, ts(1_000_000)); assert_eq!(consumer.discontinuity(), 1, "the hole moves the playhead"); - assert_eq!(consumer.restarts(), 0, "but the timeline carried on"); + assert_eq!(consumer.markers(), 0, "but the timeline carried on"); assert_eq!(consumer.read().await.unwrap().unwrap().timestamp, ts(2_000_000)); assert_eq!(consumer.discontinuity(), 2); - assert_eq!(consumer.restarts(), 1, "the marker declares a restart"); + assert_eq!(consumer.markers(), 1, "the marker declares a break"); } #[tokio::test] @@ -1351,7 +1352,7 @@ mod tests { assert_eq!( consumer.discontinuity(), 1, - "a drained marker still declares the encoder restart" + "a drained marker still declares the encoder's break" ); group0.finish().unwrap(); } diff --git a/rs/moq-mux/src/container/source.rs b/rs/moq-mux/src/container/source.rs index b3424cc4c8..ba386ed32a 100644 --- a/rs/moq-mux/src/container/source.rs +++ b/rs/moq-mux/src/container/source.rs @@ -234,7 +234,7 @@ impl ExportSource { } /// The underlying consumer's playhead generation, or 0 until the subscription resolves: - /// it counts skipped groups as well as restarts ([`Consumer::discontinuity`]). + /// it counts skipped groups as well as markers ([`Consumer::discontinuity`]). pub fn skips(&self) -> u64 { match &self.state { SourceState::Active(consumer) => consumer.discontinuity(), @@ -265,16 +265,16 @@ impl ExportSource { } } - /// How many times the publisher declared its timeline restarted, or 0 until the - /// subscription resolves. + /// How many marker groups the publisher declared, each a pause or forward break in its + /// timeline, or 0 until the subscription resolves. /// - /// See [`Consumer::restarts`]. Sample it alongside each frame returned by + /// See [`Consumer::markers`]. Sample it alongside each frame returned by /// [`poll_read`](Self::poll_read): the frame read while the counter changes is the - /// first on the new timeline, so anything anchored on the media clock has to - /// re-anchor to it. A skipped group is not a restart: the timeline carries on. - pub fn restarts(&self) -> u64 { + /// first past the break, so anything anchored on the media clock has to re-anchor to + /// it. A skipped group is not a marker: the timeline carries on. + pub fn markers(&self) -> u64 { match &self.state { - SourceState::Active(consumer) => consumer.restarts(), + SourceState::Active(consumer) => consumer.markers(), _ => 0, } } diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index d2e55e0501..d3d3314e91 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -100,9 +100,19 @@ pub struct Export { source: crate::Source, catalog: Option>, catalog_format: CatalogFormat, - /// Tracks [`Self::resume`] left on the ended broadcast, resubscribed as the - /// returned catalog lists them. + /// A snapshot arrived on `catalog` since it was subscribed. + cataloged: bool, + /// The publisher epoch of the broadcast being exported, `None` for an epochless route. + instance: Option, + /// Tracks [`Self::follow`] left on the ended broadcast, resubscribed as the returned + /// catalog lists them. stale: HashSet, + /// The PAT and PMT `version_number`, advanced by each switch to another instance so a + /// demux that caches tables by version reads the new ones. + version: VersionNumber, + /// After a switch to another instance, the PIDs whose first packet is queued flagging the + /// break. `None` for an export that never switched. + flagged: Option>, /// How long after its decode time each frame goes out. delay: Duration, /// Holds every track's frames until `delay` past their decode time, keyed by PID. @@ -182,9 +192,10 @@ struct Tally { /// A frame read from its source. struct Pending { frame: Frame, - /// How many times the source had restarted its timeline when the frame was read. + /// How many marker groups had declared a break in the track's timeline when the frame + /// was read. restart: u64, - /// How many times its playhead had jumped, restarts included ([`jitter::Arrival::skip`]). + /// How many times its playhead had jumped, markers included ([`jitter::Arrival::skip`]). skip: u64, /// The earliest the frame could have arrived: when its source was last found empty. arrived: web_async::time::Instant, @@ -211,8 +222,11 @@ struct Track { /// no later than it is read; the jitter buffer judges it on the earlier bound, so a caller /// that polls late (a sink sleeping to pace its writes) does not make it late. empty: web_async::time::Instant, - /// The source's restart counter the decode clock runs under. + /// The marker count the decode clock runs under. restart: u64, + /// The marker and skip counts of the sources the track read before this one, so the + /// counts keep rising across a resubscription. + base: (u64, u64), finished: bool, pid: u16, kind: Kind, @@ -235,6 +249,7 @@ impl Track { pending: None, empty: web_async::time::Instant::now(), restart: 0, + base: (0, 0), finished: false, pid: 0, kind, @@ -252,7 +267,7 @@ impl Track { /// strand a B-frame behind its reference or send it first. fn queue(&mut self, name: &str, pending: Pending, jitter: &mut jitter::Buffer) -> anyhow::Result<()> { if pending.restart != self.restart { - // The source may have restarted its timeline, so the decode clock restarts too. + // A marker declared a break in the timeline, so the decode clock restarts too. self.release(name, jitter, true)?; self.restart = pending.restart; self.clock = DecodeClock::default(); @@ -741,12 +756,29 @@ impl Export { /// Shared constructor. The public entry points each live on a concrete /// `Export` impl that pins `E`, so the extension is chosen by which one you call. async fn build(source: crate::Source, catalog_format: CatalogFormat) -> Result { - let catalog = source.catalog::(catalog_format).await?; - Ok(Self { - source, + let broadcast = source.broadcast().await?; + let catalog = crate::catalog::Consumer::new(&broadcast, catalog_format).await?; + Ok(Self::build_with(source, &broadcast, catalog_format, catalog)) + } + + /// Export `broadcast`, the one at `source`'s path, from its subscribed `catalog`, pinning + /// every later request for that path to its instance. + fn build_with( + source: crate::Source, + broadcast: &moq_net::broadcast::Consumer, + catalog_format: CatalogFormat, + catalog: crate::catalog::Consumer, + ) -> Self { + let instance = broadcast.info().epoch.clone(); + Self { + source: source.pinned(instance.clone()), catalog: Some(catalog), catalog_format, + cataloged: false, + instance, stale: HashSet::new(), + version: VersionNumber::default(), + flagged: None, delay: Duration::ZERO, jitter: jitter::Buffer::new(Duration::ZERO), held: None, @@ -771,7 +803,7 @@ impl Export { video_start: None, mux_rate: None, mux_rate_override: None, - }) + } } /// Pad the output with null packets to `mux_rate` bits per second, whatever the @@ -848,7 +880,10 @@ impl Export { // 1. Drain catalog updates, discovering the track layout. while let Some(catalog) = self.catalog.as_mut() { match catalog.poll_next(waiter)? { - Poll::Ready(Some(snapshot)) => self.update_catalog(snapshot)?, + Poll::Ready(Some(snapshot)) => { + self.cataloged = true; + self.update_catalog(snapshot)? + } Poll::Ready(None) => { self.catalog = None; break; @@ -929,9 +964,8 @@ impl Export { // Once every source has ended, the tail goes out late if it must: the end of the // stream is not a missed deadline. - if !self.tracks.is_empty() && self.tracks.values().all(|track| track.finished) { - self.schedule.end(); - } + self.schedule + .set_ended(!self.tracks.is_empty() && self.tracks.values().all(|track| track.finished)); // 4. Mux each frame the jitter buffer lets go (the first carries the buffered // PAT/PMT), then lay out every grid slot whose time has come ([`Self::lay_due`]). @@ -1089,8 +1123,8 @@ impl Export { } let pending = Pending { frame, - restart: track.source.restarts(), - skip: track.source.skips(), + restart: track.base.0 + track.source.markers(), + skip: track.base.1 + track.source.skips(), arrived: track.empty, read: web_async::time::Instant::now(), }; @@ -1342,7 +1376,9 @@ impl Export { /// Point each stale track this snapshot lists at the returned broadcast. /// /// The PIDs and PMT stay as announced. A track the returned catalog does not list stays - /// finished, silent on its PID. + /// finished, silent on its PID. The new subscription's playhead jumped from wherever the + /// old one stopped, so its first frame counts as a skip: late frames drop, and a leap + /// ahead opens a new generation. fn resubscribe(&mut self, catalog: &Catalog, mpegts: &catalog::Mpegts) -> anyhow::Result<()> { let budget = self.budget(); for (name, track) in self.tracks.iter_mut() { @@ -1362,6 +1398,8 @@ impl Export { continue; }; self.stale.remove(name); + track.base.0 += track.source.markers(); + track.base.1 += track.source.skips() + 1; track.source = source; track.finished = false; track.empty = web_async::time::Instant::now(); @@ -1425,39 +1463,90 @@ impl Export { self.emitted_epoch } - /// Carry on with the broadcast that replaced the one this export was reading. + pub(super) fn source(&self) -> &crate::Source { + &self.source + } + + /// The publisher epoch of the broadcast being exported, `None` for an epochless route. + pub(super) fn instance(&self) -> Option<&moq_net::Epoch> { + self.instance.as_ref() + } + + /// Carry on into `broadcast`, the one now at this export's path. /// - /// Call it once [`Source::returned`](crate::Source::returned) has resolved, after - /// [`next`](Self::next) returned `None` or an error. Whatever the ended broadcast left - /// unwritten is dropped and the program clock restarts, so the next output flags the - /// break and re-emits the PAT/PMT. The returned catalog resubscribes each track as it - /// lists it, under the program already announced; one that adds a track fails as a - /// layout change. - pub async fn resume(&mut self) -> Result<(), crate::Error> { - self.catalog = Some(self.source.catalog::(self.catalog_format).await?); + /// The same publisher instance (an equal epoch) continues the stream: the program and + /// its PSI stay as announced, and its catalog resubscribes each track as it lists it, so a + /// gap is a skip like any other. One that adds a track fails as a layout change. + /// + /// Another instance (another epoch, or an epochless route) is a full program switch: a + /// fresh export of `broadcast` with a new PMT from its catalog, and nothing carried from + /// this one but the PAT and PMT `version_number`s, both advanced. Every PID's first packet + /// flags the break with `discontinuity_indicator`, and each stream starts on a keyframe. + /// The caller decides whether to follow a replacement at all; [`Follower`](super::Follower) + /// decides from the path's announcements. + pub async fn follow(mut self, broadcast: moq_net::broadcast::Consumer) -> Result { + let catalog = crate::catalog::Consumer::new(&broadcast, self.catalog_format).await?; + self.followed(&broadcast, catalog)?; + Ok(self) + } + + pub(super) fn catalog_format(&self) -> CatalogFormat { + self.catalog_format + } + + /// Whether the catalog this export last subscribed has delivered a snapshot. + pub(super) fn cataloged(&self) -> bool { + self.cataloged + } + + /// [`Self::follow`], once `broadcast`'s catalog is subscribed. + pub(super) fn followed( + &mut self, + broadcast: &moq_net::broadcast::Consumer, + catalog: crate::catalog::Consumer, + ) -> Result<(), crate::Error> { + let same = self.instance.is_some() && broadcast.info().epoch == self.instance; + if !same || self.psi.is_none() { + let mut next = Self::build_with(self.source.clone(), broadcast, self.catalog_format, catalog); + next.replay = self.replay; + next = next.with_delay(self.delay); + if let Some(rate) = self.mux_rate_override { + next = next.with_mux_rate(rate); + } + // A switch away from a program that went out, or one still pending because this + // program never built its tables, carries to the next: its tables must read as new, + // every PID flags the break, and a pacing caller re-anchors on the new clock. Only a + // program that went out advances them again. + let emitted = self.psi.is_some(); + if (!same && emitted) || self.flagged.is_some() { + next.version = self.version; + next.flagged = Some(HashSet::new()); + next.pcr_discontinuity = true; + next.epoch = self.epoch; + next.emitted_epoch = self.emitted_epoch; + if !same && emitted { + next.version.increment(); + next.epoch += 1; + } + } + *self = next; + return Ok(()); + } + + self.catalog = Some(catalog); + self.cataloged = false; self.si_flushed = false; // A name no entry carries, so the returned catalog repoints every SI entry. for si in self.si.values_mut() { si.track.clear(); } - if self.psi.is_none() { - // Nothing announced yet: the returned catalog builds the program afresh. - self.tracks.clear(); - } // Whatever the ended broadcast left is done, including a track whose error ended - // the export: polled again, it would end this one too. + // the export: polled again, it would end this one too. Its held frames still go out. for (name, track) in self.tracks.iter_mut() { + track.release(name, &mut self.jitter, true)?; track.finished = true; - track.pending = None; - track.restart = 0; - track.clock = DecodeClock::default(); self.stale.insert(name.clone()); } - // The replacement's clock starts afresh at its own first frame. - self.jitter.clear(); - self.held = None; - self.generation = 0; - self.rewind(); Ok(()) } @@ -1468,6 +1557,10 @@ impl Export { fn rewind(&mut self) { self.epoch += 1; self.schedule.clear(); + // A break flag queued on a PID that has not gone out yet was just dropped with it. + if let Some(flagged) = &mut self.flagged { + flagged.clear(); + } self.queue.clear(); self.last_psi = None; for si in self.si.values_mut() { @@ -1505,6 +1598,17 @@ impl Export { .min() } + /// Whether `pid`'s next packet is its first since a switch to another instance, so it has + /// to flag the break. The PCR PID's first packet is always a clock packet, flagged on its + /// own ([`Self::lay`]). + fn breaks(&mut self, pid: u16) -> bool { + let pcr = self.psi.as_ref().map(|psi| psi.pcr_pid); + match &mut self.flagged { + Some(flagged) => Some(pid) != pcr && !self.counters.contains_key(&pid) && flagged.insert(pid), + None => false, + } + } + /// PID the PMT rides on: the source's original (preserved in the service record), /// or the synthesized [`PMT_PID`] for a media-only source or an invalid (zero) value. fn pmt_pid(&self) -> u16 { @@ -1634,7 +1738,7 @@ impl Export { let pat = Pat { transport_stream_id, - version_number: VersionNumber::default(), + version_number: self.version, table: vec![ProgramAssociation { program_num: program_number, program_map_pid: Pid::new(pmt_pid)?, @@ -1643,7 +1747,7 @@ impl Export { let pmt = Pmt { program_num: program_number, pcr_pid: Some(Pid::new(pcr_pid)?), - version_number: VersionNumber::default(), + version_number: self.version, program_info, es_info, }; @@ -1731,8 +1835,10 @@ impl Export { let pmt_pid = psi.pmt_pid; let pat = TsPayload::Pat(psi.pat.clone()); let pmt = TsPayload::Pmt(psi.pmt.clone()); - self.write_packet(&mut out, Pid::PAT, None, pat)?; - self.write_packet(&mut out, pmt_pid, None, pmt)?; + let flag = self.breaks(Pid::PAT).then(|| flags(true)); + self.write_packet(&mut out, Pid::PAT, flag, pat)?; + let flag = self.breaks(pmt_pid).then(|| flags(true)); + self.write_packet(&mut out, pmt_pid, flag, pmt)?; self.last_psi = Some(frame.timestamp); } @@ -1998,20 +2104,11 @@ impl Export { let mut offset = 0; let mut first = true; loop { - let adaptation = if first && unit.keyframe { - Some(AdaptationField { - discontinuity_indicator: false, - random_access_indicator: true, - es_priority_indicator: false, - pcr: None, - opcr: None, - splice_countdown: None, - transport_private_data: Vec::new(), - extension: None, - }) - } else { - None - }; + let flag = first && self.breaks(unit.pid); + let adaptation = ((first && unit.keyframe) || flag).then(|| AdaptationField { + random_access_indicator: first && unit.keyframe, + ..flags(flag) + }); let header_len = if first { 6 + optional_len } else { 0 }; let af_len = adaptation.as_ref().map(adaptation_size).unwrap_or(0); @@ -2057,9 +2154,12 @@ impl Export { let mut offset = 0; let mut first = true; loop { + let flag = (first && self.breaks(pid)).then(|| flags(true)); let payload = if first { - // pointer_field (1 byte, written by `Section`) eats one payload byte. - let take = (TsBytes::MAX_SIZE - 1).min(section.len()); + // pointer_field (1 byte, written by `Section`) eats one payload byte, and a flag + // for the break its adaptation field. + let room = TsBytes::MAX_SIZE - 1 - flag.as_ref().map_or(0, adaptation_size); + let take = room.min(section.len()); let chunk = §ion[..take]; offset = take; TsPayload::Section(Section { @@ -2073,7 +2173,7 @@ impl Export { TsPayload::Raw(TsBytes::new(chunk).map_err(anyhow::Error::msg)?) }; - self.write_packet(out, pid, None, payload)?; + self.write_packet(out, pid, flag, payload)?; first = false; if offset >= section.len() { break; @@ -2218,6 +2318,20 @@ fn stamp(nanos: u128) -> anyhow::Result { Ok(Timestamp::from_micros(micros)?) } +/// An adaptation field carrying nothing but `discontinuity_indicator`. +fn flags(discontinuity: bool) -> AdaptationField { + AdaptationField { + discontinuity_indicator: discontinuity, + random_access_indicator: false, + es_priority_indicator: false, + pcr: None, + opcr: None, + splice_countdown: None, + transport_private_data: Vec::new(), + extension: None, + } +} + /// External byte size of an adaptation field (manual mirror of the crate's /// private `external_size`); only PCR is ever set. fn adaptation_size(af: &AdaptationField) -> usize { diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 136ac32139..a9cfc5e2a6 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -6825,9 +6825,9 @@ async fn a_late_join_starts_at_the_live_edge() { } /// Publish a Legacy AAC rendition named `name`. -fn aac_rendition( +fn aac_rendition( broadcast: &mut moq_net::broadcast::Producer, - catalog: &mut crate::catalog::Producer, + catalog: &mut crate::catalog::Producer, name: &str, ) -> Producer { let track = broadcast @@ -6925,96 +6925,367 @@ async fn a_track_leaving_the_catalog_is_read_to_its_end() { assert_eq!(pes_count(&frames), 20 + 15, "every frame of both tracks went out"); } -/// A broadcast that ends and is published again carries on in the same export: the -/// returned catalog resubscribes the program's tracks, the clock break is flagged once, -/// and the PAT/PMT go out again for a receiver re-acquiring after the gap. Both ways a -/// broadcast ends, a clean finish and a drop, resume the same way. -async fn resume_after(finish: bool) { +/// Publish an empty catalog at `live` under `route`. +fn publish_live( + origin: &moq_net::origin::Producer, + route: moq_net::origin::Route, +) -> (moq_net::broadcast::Producer, crate::catalog::Producer) { + let mut broadcast = origin.publish("live", route).unwrap(); + let config = crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()); + let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap(); + (broadcast, catalog) +} + +/// Publish a Legacy H.264 rendition named `name`, described out of band. +fn h264_rendition( + broadcast: &mut moq_net::broadcast::Producer, + catalog: &mut crate::catalog::Producer, + name: &str, +) -> Producer { + let track = broadcast + .create_track(name, hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + let mut cfg = VideoConfig::new(H264 { + profile: 0x42, + constraints: 0xc0, + level: 0x1f, + inline: false, + }); + cfg.container = Container::Legacy; + cfg.description = + Some(crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap()); + catalog.modify().unwrap().video.renditions.insert(name.to_string(), cfg); + Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data)) +} + +/// Write one 25 fps video frame at `ms`, a keyframe opening a group every fifth. +fn write_h264(producer: &mut Producer, ms: u64) { + let keyframe = ms.is_multiple_of(200); + if keyframe && ms > 0 { + producer.cut(None).unwrap(); + } + let slice = if keyframe { [0x65u8; 64] } else { [0x41u8; 64] }; + producer + .write(Frame { + timestamp: Timestamp::from_millis(60_000 + ms).unwrap(), + duration: None, + payload: length_prefixed(&[&slice]), + keyframe, + }) + .unwrap(); +} + +/// Every PAT in `frames`: its version and the PMT PID it lists. +fn pats(frames: &[Frame]) -> Vec<(u8, u16)> { + let bytes: Vec = frames.iter().flat_map(|f| f.payload.iter().copied()).collect(); + let mut reader = TsPacketReader::new(Cursor::new(bytes)); + let mut out = Vec::new(); + while let Some(packet) = reader.read_ts_packet().unwrap() { + if let Some(TsPayload::Pat(pat)) = packet.payload { + out.push((pat.version_number.as_u8(), pat.table[0].program_map_pid.as_u16())); + } + } + out +} + +/// Every PMT in `frames`: its version and the stream types it lists. +fn pmts(frames: &[Frame]) -> Vec<(u8, Vec)> { + let bytes: Vec = frames.iter().flat_map(|f| f.payload.iter().copied()).collect(); + let mut reader = TsPacketReader::new(Cursor::new(bytes)); + let mut out = Vec::new(); + while let Some(packet) = reader.read_ts_packet().unwrap() { + if let Some(TsPayload::Pmt(pmt)) = packet.payload { + let types = pmt.es_info.iter().map(|es| es.stream_type).collect(); + out.push((pmt.version_number.as_u8(), types)); + } + } + out +} + +/// The PIDs whose first packet in `frames` does not flag a discontinuity. +fn unflagged(frames: &[Frame]) -> Vec { + let mut seen = std::collections::HashSet::new(); + let mut out = Vec::new(); + for packet in frames.iter().flat_map(|f| f.payload.as_chunks::<188>().0.iter()) { + let pid = u16::from(packet[1] & 0x1f) << 8 | u16::from(packet[2]); + if pid == 0x1fff || !seen.insert(pid) { + continue; + } + if !(packet[3] & 0x20 != 0 && packet[4] > 0 && packet[5] & 0x80 != 0) { + out.push(pid); + } + } + out +} + +/// The same instance back after every route went carries on in the same export, under the +/// program already announced: the PSI keeps its version, and nothing flags a break across a +/// gap the clock spans. Both ways a broadcast ends, a clean finish and a drop, continue alike. +async fn same_instance_returns(finish: bool) { let origin = crate::source::produce_origin(); let source = crate::Source::new(origin.consume(), "live"); - let publish = || { - let mut broadcast = origin.publish("live", Default::default()).unwrap(); - let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); - (broadcast, catalog) - }; + let epoch = moq_net::Epoch::mint(); + let route = || moq_net::origin::Route::default().with_epoch(epoch.clone()); - let (mut broadcast, mut catalog) = publish(); + let (mut broadcast, mut catalog) = publish_live::<()>(&origin, route()); let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); - let ended = source.broadcast().await.unwrap(); let mut export = Export::new(source.clone()) .await .unwrap() .with_delay(RECORDING_MAX_AGE) .with_replay(); + let start = tokio::time::Instant::now(); for ms in (0..200).step_by(20) { write_aac(&mut track, ms); } - let mut before = drain_frames(&mut export).await; + let mut frames = drain_frames(&mut export).await; if finish { track.finish().unwrap(); catalog.finish().unwrap(); } drop((broadcast, catalog, track)); let (rest, end) = drain_to_end(&mut export).await; - before.extend(rest); + frames.extend(rest); assert_eq!(end.is_ok(), finish, "a finish ends cleanly and a drop fails: {end:?}"); - // The old broadcast ends before the publisher comes back: a path still routed is one - // broadcast, so a publisher back before then would resume it instead. - ended.closed().await; - - // The returned catalog lists the track only in its second snapshot, and the restarted - // publisher's clock starts over. - let (mut broadcast, mut catalog) = publish(); + // Back under the same epoch a second later, its media as far on as the wall clock, its + // catalog listing the track only in its second snapshot. + tokio::time::sleep(Duration::from_secs(1)).await; + let (mut broadcast, mut catalog) = publish_live::<()>(&origin, route()); std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap()); - source.returned(&ended).await.unwrap(); - export.resume().await.unwrap(); - let mut after = drain_frames(&mut export).await; - assert!(after.is_empty(), "nothing to carry before the track is listed"); + let back = origin.consume().routed_broadcast("live").await.unwrap(); + let mut export = export.follow(back).await.unwrap(); + frames.extend(drain_frames(&mut export).await); let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); - for ms in (0..200).step_by(20) { + let resumed = start.elapsed().as_millis() as u64; + for ms in (resumed..resumed + 200).step_by(20) { write_aac(&mut track, ms); } - after.extend(drain_frames(&mut export).await); + frames.extend(drain_frames(&mut export).await); track.finish().unwrap(); catalog.finish().unwrap(); let (rest, end) = drain_to_end(&mut export).await; - after.extend(rest); + frames.extend(rest); end.unwrap(); - assert_eq!(export.discontinuity(), 1, "the return is one break"); - assert_eq!(count_discontinuity(&before), 0); - assert_eq!(count_discontinuity(&after), 1, "the break is flagged once"); - assert!(count_pid(&after, 0x0000) >= 1, "PAT re-emitted after the return"); - assert_eq!(pes_count(&after), 10, "the returned broadcast's frames all went out"); + assert_eq!(pes_count(&frames), 20, "every frame of both spans went out"); + assert_eq!(count_discontinuity(&frames), 0, "a gap the clock spans is no break"); + assert_eq!(export.discontinuity(), 0); + assert!( + pats(&frames).iter().all(|&(version, _)| version == 0), + "the PAT keeps its version" + ); + assert!( + pmts(&frames).iter().all(|(version, _)| *version == 0), + "the PMT keeps its version" + ); } #[tokio::test(start_paused = true)] -async fn resume_after_a_finish() { - resume_after(true).await; +async fn the_same_instance_continues_after_a_finish() { + same_instance_returns(true).await; } #[tokio::test(start_paused = true)] -async fn resume_after_a_drop() { - resume_after(false).await; +async fn the_same_instance_continues_after_a_drop() { + same_instance_returns(false).await; } -/// The stats count only output that was returned. A frame the muxer refuses fails the export -/// with the span before it queued but never returned, and the resume discards it. +/// A switch onto an instance that is itself replaced before it builds its tables still +/// reaches the next program as a switch: its tables carry a new version, every PID's first +/// packet flags the break, and pacing sees a new generation. #[tokio::test(start_paused = true)] -async fn export_stats_skip_output_a_failure_discards() { +async fn a_switch_carries_through_an_instance_that_never_built_its_tables() { let origin = crate::source::produce_origin(); let source = crate::Source::new(origin.consume(), "live"); - let publish = || { - let mut broadcast = origin.publish("live", Default::default()).unwrap(); - let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); - (broadcast, catalog) + let route = || moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + + let (mut old, mut old_catalog) = publish_live::<()>(&origin, route()); + let mut old_audio = aac_rendition(&mut old, &mut old_catalog, "a.aac"); + let mut export = Export::new(source.clone()) + .await + .unwrap() + .with_delay(RECORDING_MAX_AGE) + .with_replay(); + for ms in (0..200).step_by(20) { + write_aac(&mut old_audio, ms); + } + let before = drain_frames(&mut export).await; + assert!(!pats(&before).is_empty(), "the first program went out"); + + // Switched to an instance whose catalog lists nothing, so it never builds its tables... + let (_bare, _bare_catalog) = publish_live::<()>(&origin, route()); + let bare = origin.consume().request_broadcast("live", None).await.unwrap(); + let mut export = export.follow(bare).await.unwrap(); + assert!( + pats(&drain_frames(&mut export).await).is_empty(), + "no tables for an empty catalog" + ); + + // ...and then to another. + let (mut new, mut new_catalog) = publish_live::<()>(&origin, route()); + let mut new_audio = aac_rendition(&mut new, &mut new_catalog, "b.aac"); + let replacement = origin.consume().request_broadcast("live", None).await.unwrap(); + let mut export = export.follow(replacement).await.unwrap(); + for ms in (0..200).step_by(20) { + write_aac(&mut new_audio, ms); + } + let after = drain_frames(&mut export).await; + assert!(!pats(&after).is_empty(), "the last program went out"); + assert!( + pats(&after).iter().all(|&(version, _)| version >= 1), + "the PAT reads as new: {:?}", + pats(&after) + ); + assert!( + pmts(&after).iter().all(|(version, _)| *version >= 1), + "the PMT reads as new" + ); + assert_eq!(unflagged(&after), Vec::::new(), "every PID flags the break"); + assert!(export.discontinuity() >= 1, "pacing sees a new generation"); +} + +/// Following another instance is a full program switch, even while the old one stays up and +/// keeps writing: a new PMT version from the replacement's catalog, whatever its codecs and +/// tracks, every PID's first packet flagging the break, each stream starting on a keyframe, +/// and nothing of the old broadcast after it. +#[tokio::test(start_paused = true)] +async fn another_instance_is_a_full_program_switch() { + let origin = crate::source::produce_origin(); + let source = crate::Source::new(origin.consume(), "live"); + let route = || moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + + let (mut old, mut old_catalog) = publish_live::<()>(&origin, route()); + let mut old_audio = aac_rendition(&mut old, &mut old_catalog, "a.aac"); + let mut export = Export::new(source.clone()) + .await + .unwrap() + .with_delay(RECORDING_MAX_AGE) + .with_replay(); + for ms in (0..200).step_by(20) { + write_aac(&mut old_audio, ms); + } + let before = drain_frames(&mut export).await; + assert_eq!( + pmts(&before).last().map(|(_, types)| types.clone()), + Some(vec![StreamType::AdtsAac]) + ); + + let (mut new, mut new_catalog) = publish_live::<()>(&origin, route()); + let mut video = h264_rendition(&mut new, &mut new_catalog, "video.avc1"); + let mut new_audio = aac_rendition(&mut new, &mut new_catalog, "b.aac"); + let replacement = origin.consume().request_broadcast("live", None).await.unwrap(); + let mut export = export.follow(replacement).await.unwrap(); + + // The old instance carries on ten seconds ahead, where none of it may land. + for ms in (10_000..10_200).step_by(20) { + write_aac(&mut old_audio, ms); + } + for ms in (0..400).step_by(40) { + write_h264(&mut video, ms); + } + for ms in (0..400).step_by(20) { + write_aac(&mut new_audio, ms); + } + let mut after = drain_frames(&mut export).await; + video.finish().unwrap(); + new_audio.finish().unwrap(); + new_catalog.finish().unwrap(); + let (rest, end) = drain_to_end(&mut export).await; + after.extend(rest); + end.unwrap(); + + assert!(pats(&before).iter().all(|&(version, _)| version == 0)); + assert!(pmts(&before).iter().all(|(version, _)| *version == 0)); + assert!(!pats(&after).is_empty() && pats(&after).iter().all(|&(version, _)| version == 1)); + let tables = pmts(&after); + assert!(!tables.is_empty()); + for (version, types) in tables { + assert_eq!(version, 1, "the replacement's PMT is a new version"); + assert_eq!(types, [StreamType::AdtsAac, StreamType::H264]); + } + assert_eq!( + unflagged(&after), + Vec::::new(), + "every PID's first packet flags the break" + ); + assert_eq!(export.discontinuity(), 1, "a pacing caller sees the new clock"); + + let bytes: Vec = after.iter().flat_map(|f| f.payload.iter().copied()).collect(); + let (video_pts, audio_pts) = collect_pes_pts(&bytes); + assert_eq!( + video_pts.first(), + Some(&(60_000 * 90)), + "the video starts on its keyframe" + ); + assert_eq!(video_pts.len(), 10); + assert!( + audio_pts.iter().all(|&pts| pts < 60_400 * 90), + "nothing of the old broadcast after the break: {audio_pts:?}" + ); +} + +/// A switch that keeps the transport stream ID but moves the PMT advances the PAT version, +/// or a demux caching the PAT by version keeps reading the old PMT PID. +#[tokio::test(start_paused = true)] +async fn a_switch_moving_the_pmt_advances_the_pat() { + let origin = crate::source::produce_origin(); + let source = crate::Source::new(origin.consume(), "live"); + let route = || moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + let publish = |pmt_pid: u16| { + let (mut broadcast, mut catalog) = publish_live::(&origin, route()); + let track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); + catalog.modify().unwrap().ext.mpegts.program = Some(tscat::Program { + transport_stream_id: 7, + program_number: 1, + pmt_pid, + }); + (broadcast, catalog, track) }; + + let (_old, _old_catalog, mut old_track) = publish(0x100); + let mut export = Export::with_ts(source.clone(), crate::catalog::CatalogFormat::Hang) + .await + .unwrap() + .with_delay(RECORDING_MAX_AGE) + .with_replay(); + for ms in (0..200).step_by(20) { + write_aac(&mut old_track, ms); + } + let before = drain_frames(&mut export).await; + + let (_new, mut new_catalog, mut new_track) = publish(0x200); + let replacement = origin.consume().request_broadcast("live", None).await.unwrap(); + let mut export = export.follow(replacement).await.unwrap(); + for ms in (0..200).step_by(20) { + write_aac(&mut new_track, ms); + } + let mut after = drain_frames(&mut export).await; + new_track.finish().unwrap(); + new_catalog.finish().unwrap(); + let (rest, end) = drain_to_end(&mut export).await; + after.extend(rest); + end.unwrap(); + + assert!(!pats(&before).is_empty() && pats(&before).iter().all(|&pat| pat == (0, 0x100))); + let after = pats(&after); + assert!(!after.is_empty()); + assert!(after.iter().all(|&pat| pat == (1, 0x200)), "{after:?}"); +} + +/// The stats count only output that was returned. A frame the muxer refuses fails the export +/// with the span before it queued but not yet returned, and the same instance coming back +/// carries on with it. +#[tokio::test(start_paused = true)] +async fn export_stats_count_only_returned_output() { + let origin = crate::source::produce_origin(); + let source = crate::Source::new(origin.consume(), "live"); + let epoch = moq_net::Epoch::mint(); + let route = || moq_net::origin::Route::default().with_epoch(epoch.clone()); let units = |stats: stats::Export| stats.streams.values().map(|row| row.units).sum::() as usize; - let (mut broadcast, mut catalog) = publish(); + let (mut broadcast, mut catalog) = publish_live::<()>(&origin, route()); let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); - let ended = source.broadcast().await.unwrap(); let mut export = Export::new(source.clone()).await.unwrap().with_delay(RECORDING_MAX_AGE); for ms in (0..200).step_by(20) { write_aac(&mut track, ms); @@ -7035,14 +7306,11 @@ async fn export_stats_skip_output_a_failure_discards() { assert_eq!(units(export.stats()), pes_count(&frames)); drop((broadcast, catalog, track)); - // The old broadcast ends before the publisher comes back: a path still routed is one - // broadcast, so a publisher back before then would resume it instead. - ended.closed().await; - let (mut broadcast, mut catalog) = publish(); + let (mut broadcast, mut catalog) = publish_live::<()>(&origin, route()); let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac"); - source.returned(&ended).await.unwrap(); - export.resume().await.unwrap(); - for ms in (0..200).step_by(20) { + let back = origin.consume().routed_broadcast("live").await.unwrap(); + let mut export = export.follow(back).await.unwrap(); + for ms in (1_000..1_200).step_by(20) { write_aac(&mut track, ms); } frames.extend(drain_frames(&mut export).await); @@ -7051,11 +7319,7 @@ async fn export_stats_skip_output_a_failure_discards() { let (rest, end) = drain_to_end(&mut export).await; frames.extend(rest); end.unwrap(); - assert_eq!( - units(export.stats()), - pes_count(&frames), - "the discarded span never counts" - ); + assert_eq!(units(export.stats()), pes_count(&frames)); } /// Export 25 fps video and two AAC tracks for [`TICKS`] video frames, the video and the first @@ -7234,3 +7498,457 @@ async fn export_stats_count_a_gap_longer_than_the_backfill() { let quiet = running.quiet.expect("the output carries a PCR"); assert!(quiet < Duration::from_millis(200), "{quiet:?}"); } + +/// Publish a catalog-only broadcast at `live`, for a test of what follows its end. +fn publish_bare( + origin: &moq_net::origin::Producer, + epoch: Option<&moq_net::Epoch>, +) -> (moq_net::broadcast::Producer, crate::catalog::Producer<()>) { + let route = moq_net::origin::Route::default(); + publish_live::<()>( + origin, + epoch.map_or(route.clone(), |epoch| route.with_epoch(epoch.clone())), + ) +} + +/// Follow the broadcast at `live` with a fresh export of it. +async fn follower(origin: &moq_net::origin::Producer) -> super::Follower { + let source = crate::Source::new(origin.consume(), "live"); + super::Follower::new(Export::new(source).await.unwrap()).unwrap() +} + +/// End a catalog-only broadcast cleanly. +fn finish((broadcast, mut catalog): (moq_net::broadcast::Producer, crate::catalog::Producer<()>)) { + catalog.finish().unwrap(); + 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. +#[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" + ); + + 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)); +} + +/// A replacement followed with stitching that never serves its catalog gives up at the +/// linger, rather than waiting on the catalog past it. +#[tokio::test(start_paused = true)] +async fn a_follower_return_without_a_catalog_expires_with_the_linger() { + 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(); + finish(first); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "lingering" + ); + + // Replaced, but the replacement's catalog request is never answered. + let second = origin + .publish( + "live", + 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); +} + +/// 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. +#[tokio::test(start_paused = true)] +async fn a_follower_return_whose_catalog_never_snapshots_expires_with_the_linger() { + 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(); + finish(first); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "lingering" + ); + + 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()) + .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); +} + +/// 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. +#[tokio::test(start_paused = true)] +async fn a_follower_keeps_lingering_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 start = tokio::time::Instant::now(); + finish(first); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "lingering" + ); + + // Back, but its catalog request is never answered before it goes again. + let second = origin + .publish("live", moq_net::origin::Route::default().with_epoch(epoch)) + .unwrap(); + let unanswered = second.dynamic(); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "resolving the return" + ); + drop((second, unanswered)); + + let end = tokio::time::timeout(linger * 3, follower.next()) + .await + .expect("the linger bounds the wait"); + assert!(matches!(end, Ok(None)), "the original clean end: {end:?}"); + assert_eq!(start.elapsed(), linger); +} + +/// A stitch whose replacement goes before it resolves is no switch yet: the export stays on +/// its own instance and follows the next replacement instead. +#[tokio::test(start_paused = true)] +async fn a_follower_stitches_onto_the_next_replacement_when_one_goes_before_it_resolves() { + let origin = crate::source::produce_origin(); + let route = || moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + let first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let mut follower = follower(&origin).await.with_stitch(true); + + // Replaced, but the replacement's catalog request is never answered before it goes. + let second = origin.publish("live", route()).unwrap(); + let unanswered = second.dynamic(); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "resolving the replacement" + ); + drop((second, unanswered)); + let third = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "following the next replacement" + ); + + // Once the first instance is gone too, the third one's end ends the export. + finish(first); + finish(third); + let end = tokio::time::timeout(Duration::from_secs(10), follower.next()) + .await + .expect("the third instance's end ends the export"); + assert!(matches!(end, Ok(None)), "{end:?}"); +} + +/// A replacement that stays announced but refuses its catalog fails a stitch loudly, rather +/// than leaving the export on the instance it replaced. +#[tokio::test(start_paused = true)] +async fn a_follower_fails_a_stitch_onto_a_replacement_that_refuses_its_catalog() { + 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); + + let _second = origin + .publish( + "live", + moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), + ) + .unwrap(); + let end = tokio::time::timeout(Duration::from_secs(10), follower.next()) + .await + .expect("the refusal ends the export"); + assert!(end.is_err(), "the refusal fails the export: {end:?}"); +} + +/// An export whose route went before the follower was built has already lost it, so the same +/// instance announcing again is a return to follow. +#[tokio::test(start_paused = true)] +async fn a_follower_built_after_its_route_went_follows_the_return() { + let origin = crate::source::produce_origin(); + let epoch = moq_net::Epoch::mint(); + let first = publish_bare(&origin, Some(&epoch)); + let source = crate::Source::new(origin.consume(), "live"); + let export = Export::new(source).await.unwrap(); + let start = tokio::time::Instant::now(); + 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); + assert!( + tokio::time::timeout(Duration::from_secs(2), follower.next()) + .await + .is_err(), + "lingering" + ); + let second = publish_bare(&origin, Some(&epoch)); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "carried on" + ); + + // Carried on into the return, so its end starts the linger over. + finish(second); + let end = follower.next().await; + assert!(matches!(end, Ok(None)), "{end:?}"); + assert_eq!(start.elapsed(), Duration::from_secs(4) + linger); +} + +/// The export's own instance winning the path back from a more specific route that went is no +/// replacement: its end ends the export cleanly, without stitching. +#[tokio::test(start_paused = true)] +async fn a_follower_takes_its_own_instance_winning_back_for_no_replacement() { + let origin = crate::source::produce_origin(); + let epoch = moq_net::Epoch::mint(); + let route = |epoch| moq_net::origin::Route::default().with_epoch(epoch); + + // The export resolves through a prefix claim serving its own instance. + let mut info = moq_net::broadcast::Info::new(); + info.epoch = Some(epoch.clone()); + let mut served = info.produce(); + let mut catalog = crate::catalog::Producer::<()>::new(&mut served, Default::default()).unwrap(); + let pool = origin.dynamic("pool", route(epoch)).unwrap(); + let consumer = served.consume(); + tokio::spawn(async move { + while let Ok(request) = pool.requested_broadcast().await { + request.accept(consumer.clone()); + } + }); + let source = crate::Source::new(origin.consume(), "pool/job"); + let mut follower = super::Follower::new(Export::new(source).await.unwrap()).unwrap(); + + // A more specific route takes the path for a while, then goes. + let exact = origin.publish("pool/job", route(moq_net::Epoch::mint())).unwrap(); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "still on its own instance" + ); + drop(exact); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "its own instance wins back" + ); + + catalog.finish().unwrap(); + let end = tokio::time::timeout(Duration::from_secs(10), follower.next()) + .await + .expect("its own end ends the export"); + assert!(matches!(end, Ok(None)), "{end:?}"); +} + +/// Without stitching, another instance taking the path fails the follower once the export +/// ends, however long the linger. +#[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)); + finish(first); + let _second = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + + let start = tokio::time::Instant::now(); + let err = follower.next().await.expect_err("a replacement fails the follower"); + assert!(matches!(err, crate::Error::Replaced(_)), "{err}"); + assert!(start.elapsed() < Duration::from_secs(1), "no wait for a return"); +} + +/// An epochless route has no instance to match, so even its own return is a replacement. +#[tokio::test(start_paused = true)] +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)); + finish(first); + let _second = publish_bare(&origin, None); + + let err = follower + .next() + .await + .expect_err("an epochless return fails the follower"); + 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. +#[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 start = tokio::time::Instant::now(); + finish(first); + assert!( + tokio::time::timeout(Duration::from_secs(2), follower.next()) + .await + .is_err(), + "lingering" + ); + + let second = publish_bare(&origin, Some(&epoch)); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "carried on" + ); + finish(second); + let end = follower.next().await.unwrap(); + assert!(end.is_none()); + assert_eq!(start.elapsed(), Duration::from_secs(3) + linger); +} + +/// Pull frames from `follower` until it ends, or until nothing more comes within the drain. +async fn drain_follower(follower: &mut super::Follower) -> (Vec, Option>) { + let mut out = Vec::new(); + loop { + match tokio::time::timeout(DRAIN, follower.next()).await { + Ok(Ok(Some(frame))) => out.push(frame), + Ok(Ok(None)) => return (out, Some(Ok(()))), + Ok(Err(err)) => return (out, Some(Err(err))), + Err(_) => return (out, None), + } + } +} + +/// The exact route going and a covering prefix of the same epoch taking over after a gap that +/// ended the export's request, with the follower unpolled across it, is its own instance +/// returning: the export carries on through the prefix under the program already announced, +/// with no break flagged. +#[tokio::test(start_paused = true)] +async fn a_follower_continues_through_a_same_epoch_handoff_after_a_gap() { + let origin = crate::source::produce_origin(); + let epoch = moq_net::Epoch::mint(); + let route = moq_net::origin::Route::default().with_epoch(epoch.clone()); + let mut exact = origin.publish("pool/job", route.clone()).unwrap(); + let mut catalog = crate::catalog::Producer::<()>::new(&mut exact, Default::default()).unwrap(); + let mut track = aac_rendition(&mut exact, &mut catalog, "a.aac"); + let source = crate::Source::new(origin.consume(), "pool/job"); + let export = Export::new(source) + .await + .unwrap() + .with_delay(RECORDING_MAX_AGE) + .with_replay(); + let mut follower = super::Follower::new(export) + .unwrap() + .with_linger(Duration::from_secs(10)); + let start = tokio::time::Instant::now(); + for ms in (0..200).step_by(20) { + write_aac(&mut track, ms); + } + let (mut frames, end) = drain_follower(&mut follower).await; + assert!(end.is_none(), "still exporting: {end:?}"); + + // 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)); + tokio::time::sleep(Duration::from_secs(1)).await; + let mut info = moq_net::broadcast::Info::new(); + info.epoch = Some(epoch); + let mut served = info.produce(); + let mut catalog = crate::catalog::Producer::<()>::new(&mut served, Default::default()).unwrap(); + let mut track = aac_rendition(&mut served, &mut catalog, "a.aac"); + let pool = origin.dynamic("pool", route).unwrap(); + let consumer = served.consume(); + tokio::spawn(async move { + while let Ok(request) = pool.requested_broadcast().await { + request.accept(consumer.clone()); + } + }); + let resumed = start.elapsed().as_millis() as u64; + for ms in (resumed..resumed + 200).step_by(20) { + write_aac(&mut track, ms); + } + track.finish().unwrap(); + catalog.finish().unwrap(); + let (rest, end) = drain_follower(&mut follower).await; + frames.extend(rest); + assert!( + matches!(end, Some(Ok(()))), + "the prefix's broadcast ends the export: {end:?}" + ); + + assert_eq!(pes_count(&frames), 20, "every frame of both routes went out"); + assert_eq!(count_discontinuity(&frames), 0, "the same instance is no break"); + assert_eq!(follower.export().discontinuity(), 0); + assert!( + pats(&frames).iter().all(|&(version, _)| version == 0), + "the PAT keeps its version" + ); + assert!( + pmts(&frames).iter().all(|(version, _)| *version == 0), + "the PMT keeps its version" + ); +} + +/// A broader prefix starting, restarting, and ending while the exact route still serves the +/// path changes nothing: the request resolves through the exact route, so none of it is a +/// replacement, and the export ends cleanly with its own broadcast. +#[tokio::test(start_paused = true)] +async fn a_follower_ignores_a_covering_prefix_while_the_exact_route_serves() { + let origin = crate::source::produce_origin(); + let route = |epoch| moq_net::origin::Route::default().with_epoch(epoch); + let mut exact = origin.publish("pool/job", route(moq_net::Epoch::mint())).unwrap(); + let catalog = crate::catalog::Producer::<()>::new(&mut exact, Default::default()).unwrap(); + let source = crate::Source::new(origin.consume(), "pool/job"); + let mut follower = super::Follower::new(Export::new(source).await.unwrap()).unwrap(); + let quiet = Duration::from_secs(1); + + let pool = origin.dynamic("pool", route(moq_net::Epoch::mint())).unwrap(); + assert!(tokio::time::timeout(quiet, follower.next()).await.is_err(), "a start"); + drop(pool); + let pool = origin.dynamic("pool", route(moq_net::Epoch::mint())).unwrap(); + assert!(tokio::time::timeout(quiet, follower.next()).await.is_err(), "a restart"); + drop(pool); + assert!(tokio::time::timeout(quiet, follower.next()).await.is_err(), "an end"); + + finish((exact, catalog)); + let end = follower.next().await; + assert!(matches!(end, Ok(None)), "no replacement: {end:?}"); +} diff --git a/rs/moq-mux/src/container/ts/follower.rs b/rs/moq-mux/src/container/ts/follower.rs new file mode 100644 index 0000000000..c13b0a387e --- /dev/null +++ b/rs/moq-mux/src/container/ts/follower.rs @@ -0,0 +1,394 @@ +//! [`Follower`]: an [`Export`] driven by its path's announcements, as a player is. + +use std::future::Future; +use std::pin::Pin; +use std::task::Poll; +use std::time::Duration; + +use web_async::time::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)>; + +#[cfg(not(target_family = "wasm"))] +type Resolving = Pin> + Send>>; +#[cfg(target_family = "wasm")] +type Resolving = Pin>>>; + +/// What the announcements say serves the exported path. +#[derive(Clone, Debug, PartialEq, Eq)] +enum Serving { + /// Nothing: every route went. + Gone, + /// The publisher instance being exported. + Ours, + /// Another instance: a newer epoch, or any return of an epochless route. + Other(Option), +} + +enum State { + /// Exporting. A return carries how the export ended and when the linger runs out until its + /// catalog delivers a snapshot, since a subscribed catalog that never answers is no return. + Running { + settling: Option<(crate::Result<()>, Instant)>, + }, + /// The export ended with `end`, waiting for the announcements to say what follows. + Settling { + 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, + /// How the export ended and when the linger runs out, or `None` for a stitch mid-stream. + settling: Option<(crate::Result<()>, Instant)>, + }, + /// Nothing follows. + Done, +} + +/// Carries an [`Export`] across its broadcast's returns, following the path's announcements. +/// +/// The same publisher instance (the same epoch) returning within the [linger](Self::with_linger) +/// 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. +/// +/// 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, + 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, + /// The origin closed, so nothing more is announced. + closed: bool, + linger: Duration, + stitch: bool, + timer: Option>>, +} + +impl Follower { + /// Follow the path `export` reads, from the broadcast it resolved. + /// + /// 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 mut follower = Self { + export, + state: State::Running { settling: None }, + announced, + path, + epoch, + serving: Serving::Ours, + ended: false, + started: false, + closed: false, + linger: 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; + } + 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, up to the returned catalog's first snapshot. + pub fn with_linger(mut self, linger: Duration) -> Self { + self.linger = linger; + self + } + + /// Follow another publisher instance taking the path, as a full program switch (see + /// [`Export::follow`]). Defaults to `false`. + pub fn with_stitch(mut self, stitch: bool) -> Self { + self.stitch = stitch; + self + } + + /// The export being carried on. + pub fn export(&self) -> &Export { + &self.export + } + + /// Get the next muxed frame, or `None` once the broadcast ends with nothing to follow. + pub async fn next(&mut self) -> crate::Result> { + kio::wait(|waiter| self.poll_next(waiter)).await + } + + /// Poll for the next muxed frame, registering `waiter` when there is none yet. + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { + let next = self.poll_state(waiter); + if matches!(next, Poll::Ready(Ok(None) | Err(_))) { + self.state = State::Done; + } + next + } + + fn poll_state(&mut self, waiter: &kio::Waiter) -> Poll>> { + loop { + match std::mem::replace(&mut self.state, State::Done) { + State::Done => return Poll::Ready(Ok(None)), + State::Running { mut settling } => { + 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(), + settling, + }; + continue; + } + let next = self.export.poll_next(waiter); + if self.export.cataloged() { + settling = None; + } + let end = match next { + Poll::Ready(Ok(Some(frame))) => { + self.state = State::Running { settling }; + return Poll::Ready(Ok(Some(frame))); + } + Poll::Ready(Ok(None)) => Ok(()), + Poll::Ready(Err(err)) => Err(err), + Poll::Pending => { + if let Some((end, deadline)) = &settling + && self.elapsed(waiter, *deadline) + { + tracing::info!(path = %self.path, linger = ?self.linger, "broadcast did not return"); + return Poll::Ready(end.clone().map(|()| None)); + } + self.state = State::Running { settling }; + return Poll::Pending; + } + }; + let now = Instant::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); + } + self.state = State::Settling { + end, + grace, + deadline: now + self.linger, + }; + } + State::Settling { + end, + mut grace, + deadline, + } => { + self.poll_announced(waiter); + if grace.is_some() && self.serving != Serving::Ours { + 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 { + Serving::Ours => self.ended, + Serving::Other(_) if self.stitch => true, + Serving::Other(_) => return Poll::Ready(Err(crate::Error::Replaced(self.path.to_string()))), + Serving::Gone => false, + }; + if follow { + self.state = State::Following { + resolving: self.resolve(), + settling: Some((end, deadline)), + }; + continue; + } + if self.closed || self.elapsed(waiter, deadline) { + tracing::info!(path = %self.path, linger = ?self.linger, "broadcast did not return"); + return Poll::Ready(end.map(|()| None)); + } + self.state = State::Settling { end, grace, deadline }; + return Poll::Pending; + } + State::Following { + mut resolving, + settling, + } => { + // Announcements wait until the export is on the instance they describe. + if let Poll::Ready(resolved) = waiter.poll_future(resolving.as_mut()) { + let (broadcast, catalog) = match resolved { + Ok(resolved) => resolved, + // An instance that went before it resolved is no return or switch yet. + // Until the next announcement says otherwise, nothing serves the path: + // a stitch stays on the export it has, and a return keeps lingering. + // A refusal from an instance still up fails loud instead. + Err(err) => { + if !matches!( + err, + crate::Error::Moq(moq_net::Error::Unroutable | moq_net::Error::Dropped) + ) { + return Poll::Ready(Err(err)); + } + tracing::warn!(path = %self.path, %err, "broadcast went before it resolved, still waiting"); + self.serving = Serving::Gone; + self.state = match settling { + Some((end, deadline)) => State::Settling { + end, + grace: None, + deadline, + }, + None => State::Running { settling: None }, + }; + continue; + } + }; + self.export.followed(&broadcast, catalog)?; + 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 { settling }; + continue; + } + if let Some((end, deadline)) = &settling + && self.elapsed(waiter, *deadline) + { + tracing::info!(path = %self.path, linger = ?self.linger, "broadcast did not return"); + return Poll::Ready(end.clone().map(|()| None)); + } + self.state = State::Following { resolving, settling }; + return Poll::Pending; + } + } + } + } + + /// Apply every announcement on hand. + fn poll_announced(&mut self, waiter: &kio::Waiter) { + while !self.closed { + match self.announced.poll_next(waiter) { + Poll::Ready(Some(event)) => self.apply(event), + Poll::Ready(None) => self.closed = true, + Poll::Pending => return, + } + } + } + + fn apply(&mut self, event: moq_net::announce::Event) { + 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, + }; + } + + /// 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)) + }) + } + + fn lingering(&self, end: &crate::Result<()>) { + if self.linger.is_zero() { + return; + } + match end { + Ok(()) => { + tracing::info!(path = %self.path, linger = ?self.linger, "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") + } + } + } + + /// Whether `at` has come, registering `waiter` to wake then if not. + fn elapsed(&mut self, waiter: &kio::Waiter, at: Instant) -> bool { + if Instant::now() >= at { + return true; + } + let timer = self + .timer + .get_or_insert_with(|| Box::pin(web_async::time::sleep_until(at))); + if timer.deadline() != at { + timer.as_mut().reset(at); + } + waiter.poll_future(timer.as_mut()).is_ready() + } +} diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index 02b8459b4a..6def6e0bb8 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/rs/moq-mux/src/container/ts/mod.rs @@ -24,6 +24,7 @@ mod adts; mod export; +mod follower; mod health; mod import; mod mux_rate; @@ -39,6 +40,7 @@ mod catalog; pub use catalog::{Catalog, Descriptor, Ext, Framing, Mpegts, Program, SiEntry, Track, Verbatim}; pub use export::Export; +pub use follower::Follower; pub use import::*; pub use passthrough::Passthrough; pub use programs::Programs; diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index b583bc77b1..be6349c4fe 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -262,9 +262,10 @@ impl Schedule { } } - /// Every source has ended: what is queued still goes out, late if it must. - pub fn end(&mut self) { - self.ended = true; + /// Whether every source has ended, so what is queued still goes out, late if it must. + /// A source that resumes (a same-instance return) holds its deadlines again. + pub fn set_ended(&mut self, ended: bool) { + self.ended = ended; } /// Pad to `rate` bits per second from the next slot on, or stop padding. @@ -604,7 +605,7 @@ mod tests { schedule.set_rate(Some(RATE)); schedule.set_buffer(1, VIDEO); schedule.push(1, ms(1_000), unit(1, 200), true); - schedule.end(); + schedule.set_ended(true); let sent: usize = drain(&mut schedule) .into_iter() .map(|(_, per_pid, _)| per_pid.get(&1).copied().unwrap_or(0)) @@ -625,7 +626,7 @@ mod tests { }; schedule.set_buffer(1, small); schedule.push(1, ms(1_000), unit(1, 11), true); - schedule.end(); + schedule.set_ended(true); let mut sent = 0; for _ in 0..100 { let Some(slot) = schedule.next(None).unwrap() else { diff --git a/rs/moq-mux/src/error.rs b/rs/moq-mux/src/error.rs index 92554636a0..159b569bcc 100644 --- a/rs/moq-mux/src/error.rs +++ b/rs/moq-mux/src/error.rs @@ -252,6 +252,10 @@ pub enum Error { /// A capture instant is ahead of the broadcast clock's now, or before its epoch. #[error("capture time is outside the broadcast clock")] InvalidCapture, + + /// Another publisher instance replaced a followed broadcast, and stitching was off. + #[error("another publisher instance replaced broadcast `{0}`; enable stitch to follow it as a program switch")] + Replaced(String), } impl Error { diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index b00c7e5861..5f4fa61a94 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -732,19 +732,6 @@ impl Buffer { pub fn out_of_tolerance(&self) -> u64 { self.out_of_tolerance } - - /// Drop every queued frame and the clock, so the next frame starts it afresh. - pub fn clear(&mut self) { - self.clock = None; - self.previous = None; - self.leads.clear(); - self.acquire = None; - self.tracks.clear(); - self.horizon = None; - self.released = false; - self.steer = Steer::default(); - self.timer = None; - } } /// `nanos` after `at`, or before it if negative, if an instant holds it. @@ -1453,19 +1440,4 @@ mod tests { assert_eq!(due(&mut buffer), ["a40"]); assert_eq!(buffer.dropped(), 0); } - - #[tokio::test(start_paused = true)] - async fn clear_starts_a_fresh_clock() { - let start = Instant::now(); - let mut buffer = Buffer::new(DELAY); - buffer.push(1, arrival(start, 5_000, "old")).unwrap(); - buffer.clear(); - assert!(buffer.is_empty()); - - tokio::time::advance(Duration::from_secs(1)).await; - let now = Instant::now(); - assert_eq!(buffer.push(1, arrival(now, 0, "new")).unwrap(), Push::Queued); - tokio::time::advance(DELAY).await; - assert_eq!(due(&mut buffer), ["new"]); - } } diff --git a/rs/moq-mux/src/source.rs b/rs/moq-mux/src/source.rs index 85d45d4f4b..f128b6beec 100644 --- a/rs/moq-mux/src/source.rs +++ b/rs/moq-mux/src/source.rs @@ -24,6 +24,9 @@ use moq_net::AsPath; pub struct Source { origin: moq_net::origin::Consumer, path: moq_net::PathOwned, + /// The publisher instance requests for this path are pinned to, once an export resolved + /// it: a request never lands on a replacement, which would splice two broadcasts. + epoch: Option, } impl Source { @@ -38,9 +41,25 @@ impl Source { Self { origin, path: path.as_path().to_owned(), + epoch: None, } } + /// This source with every request for its own path pinned to `epoch`, the instance an + /// export resolved. `None` (an epochless route) leaves requests unpinned. + pub(crate) fn pinned(mut self, epoch: Option) -> Self { + self.epoch = epoch; + self + } + + pub(crate) fn origin(&self) -> &moq_net::origin::Consumer { + &self.origin + } + + pub(crate) fn path(&self) -> &moq_net::PathOwned { + &self.path + } + /// Resolve and subscribe to the catalog broadcast (the one at this source's path). pub async fn broadcast(&self) -> crate::Result { Ok(self.request_catalog().await?) @@ -59,18 +78,15 @@ impl Source { crate::catalog::Consumer::new(&broadcast, format).await } - /// Wait for the broadcast to come back after `ended`, the one an export was reading, closes. - /// - /// A broadcast that stays up never returns, so a clean catalog end that leaves it announced - /// waits here until the caller's own deadline. Bound the wait with a timeout. - pub async fn returned(&self, ended: &moq_net::broadcast::Consumer) -> crate::Result { - ended.closed().await; - Ok(self.origin.routed_broadcast(&self.path).await?) - } - /// Begin resolving the catalog broadcast (the one at this source's path). pub(crate) fn request_catalog(&self) -> kio::Pending { - self.origin.request_broadcast(&self.path, None) + self.request_path(&self.path) + } + + /// Begin resolving `path`, pinned to this source's epoch when it names its own broadcast. + fn request_path(&self, path: &moq_net::PathOwned) -> kio::Pending { + let epoch = self.epoch.clone().filter(|_| *path == self.path); + self.origin.request_broadcast(path, epoch) } /// Resolve a rendition's optional broadcast reference to an origin path. @@ -110,7 +126,7 @@ impl Source { &self, rel: Option<&moq_net::path::Relative<'_>>, ) -> crate::Result> { - Ok(self.origin.request_broadcast(&self.target(rel)?, None)) + Ok(self.request_path(&self.target(rel)?)) } /// The skipping counterpart to [`Self::request`], returning `None` when `rel` walks above @@ -123,7 +139,7 @@ impl Source { &self, rel: Option<&moq_net::path::Relative<'_>>, ) -> Option> { - Some(self.origin.request_broadcast(&self.resolve_reference(rel)?, None)) + Some(self.request_path(&self.resolve_reference(rel)?)) } /// Remove renditions whose broadcast reference escapes above the origin root. @@ -382,34 +398,33 @@ mod tests { assert!(binding.broadcast().await.is_err()); } + /// A source pinned to an epoch never resolves its path to a replacement, so an export's + /// later requests cannot splice another instance into it. Other paths stay unpinned. #[tokio::test] - async fn returned_waits_for_a_new_broadcast_after_the_old_one_ends() { + async fn a_pinned_source_refuses_a_replacement() { let origin = produce_origin(); - let first = origin.publish("live", Default::default()).unwrap(); + let first = moq_net::Epoch::mint(); + let _old = origin + .publish("live", moq_net::origin::Route::default().with_epoch(first.clone())) + .unwrap(); + let _sibling = origin.publish("sibling", Default::default()).unwrap(); settle().await; - let source = Source::new(origin.consume(), "live"); - let ended = source.broadcast().await.unwrap(); - - let returned = source.returned(&ended); - tokio::pin!(returned); - tokio::select! { - biased; - _ = &mut returned => panic!("the broadcast is still up"), - _ = settle() => {} - } - - // Gone, and nothing serves the path yet. - drop(first); - tokio::select! { - biased; - _ = &mut returned => panic!("nothing has published the path again"), - _ = settle() => {} - } - - let _second = origin.publish("live", Default::default()).unwrap(); - let back = returned.await.unwrap(); - assert!(!back.is_closed()); - assert!(!back.is_clone(&ended)); + let source = Source::new(origin.consume(), "live").pinned(Some(first)); + source.broadcast().await.expect("the pinned instance resolves"); + + let _new = origin + .publish( + "live", + moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), + ) + .unwrap(); + settle().await; + assert!(source.broadcast().await.is_err(), "the replacement is refused"); + let sibling = Relative::new("./sibling"); + source + .resolve(Some(&sibling)) + .await + .expect("another path is not pinned"); } #[tokio::test] diff --git a/rs/moq-srt/src/dial.rs b/rs/moq-srt/src/dial.rs index 2367a85784..32f6520495 100644 --- a/rs/moq-srt/src/dial.rs +++ b/rs/moq-srt/src/dial.rs @@ -60,6 +60,14 @@ pub struct Client { /// The programs of a multi-program remote [`pull`] publishes, or `None` to refuse one. program: Option, + + /// How long [`publish`] waits for the same publisher instance to come back once its + /// broadcast ends. [`publish`] only. + linger: Duration, + + /// Whether [`publish`] follows another instance replacing its broadcast, as a program + /// switch on the same connection. [`publish`] only. + stitch: bool, } impl Client { @@ -73,6 +81,8 @@ impl Client { max_age: None, bandwidth: moq_net::bandwidth::Allocator::unlimited(), program: None, + linger: Duration::ZERO, + stitch: false, } } @@ -101,11 +111,32 @@ impl Client { self } + /// Wait up to `linger` 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. + pub fn with_linger(mut self, linger: Duration) -> Self { + self.linger = linger; + self + } + + /// Follow another publisher instance replacing the broadcast [`publish`](Self::publish) + /// reads, as a full program switch on the same connection. Off by default, which ends the + /// push with [`moq_mux::Error::Replaced`]. + pub fn with_stitch(mut self, stitch: bool) -> Self { + self.stitch = stitch; + self + } + /// Push a MoQ broadcast out to the remote as MPEG-TS until the broadcast ends. pub async fn publish(&self, origin: &origin::Consumer, path: impl moq_net::AsPath) -> Result<()> { let path = path.as_path(); let socket = self.call(Mode::Publish).await?; - serve_subscribe(origin, path.as_str(), socket, self.latency).await + let options = crate::ts::Options { + latency: self.latency, + linger: self.linger, + stitch: self.stitch, + }; + serve_subscribe(origin, path.as_str(), socket, options).await } /// Pull a remote MPEG-TS stream into `origin` at `path` until the remote ends. diff --git a/rs/moq-srt/src/listen.rs b/rs/moq-srt/src/listen.rs index 5cc7f5476e..6630c881b7 100644 --- a/rs/moq-srt/src/listen.rs +++ b/rs/moq-srt/src/listen.rs @@ -69,6 +69,16 @@ pub struct Config { /// multiplex. To choose per path, drive [`Server`] and call /// [`Publish::with_program`](crate::Publish::with_program) on each request. pub program: Option, + + /// How long an egress waits for the same publisher instance to come back once its + /// broadcast ends, carrying on with the same stream. Zero (the default) closes the SRT + /// stream at the end. Only affects egress (`m=request`). + pub linger: Duration, + + /// Whether an egress follows another publisher instance replacing its broadcast, as a full + /// program switch on the same SRT connection. Off by default, which ends the stream with + /// [`moq_mux::Error::Replaced`]. Only affects egress (`m=request`). + pub stitch: bool, } impl Default for Config { @@ -79,6 +89,8 @@ impl Default for Config { latency: crate::server::DEFAULT_LATENCY, max_age: None, program: None, + linger: Duration::ZERO, + stitch: false, } } } @@ -117,6 +129,7 @@ pub async fn run(origin: origin::Producer, config: Config) -> Result<()> { let prefix = config.prefix; let max_age = config.max_age; let program = config.program; + let (linger, stitch) = (config.linger, config.stitch); while let Some(request) = server.accept().await { let prefix = prefix.clone(); @@ -156,6 +169,7 @@ pub async fn run(origin: origin::Producer, config: Config) -> Result<()> { tokio::spawn(async move { let peer = subscribe.peer(); let path = prefix.join(subscribe.resource()); + let subscribe = subscribe.with_linger(linger).with_stitch(stitch); if let Err(err) = subscribe.accept(&consumer, &path).await { tracing::warn!(%peer, %path, %err, "SRT request ended with error"); } else { diff --git a/rs/moq-srt/src/server.rs b/rs/moq-srt/src/server.rs index e51667dc19..08de535eff 100644 --- a/rs/moq-srt/src/server.rs +++ b/rs/moq-srt/src/server.rs @@ -134,10 +134,11 @@ struct Egress { impl Egress { /// Pace one muxer frame, returning the SRT payloads it completed. /// - /// `discontinuity` is the muxer's counter for `frame`. A change means the - /// publisher rewound and the muxer restarted its program clock, so the anchor - /// this pacer holds now belongs to a timeline that no longer exists: mapping the - /// new generation through it puts every frame of the rewound span before the + /// `discontinuity` is the muxer's counter for `frame`. A change means the muxer + /// restarted its program clock (a marker breaking the timeline, or a switch to another + /// publisher instance, whose timeline starts wherever its own does), so the anchor + /// this pacer holds belongs to a timeline that no longer applies: mapping a new + /// generation that starts earlier through it puts every frame of that span before the /// first packet, where the floor collapses all of them onto one instant and the /// receiver sees the program stop until the media climbs back. Make the frame /// the live edge instead, and pace the rest of the generation off it. @@ -249,6 +250,8 @@ impl Server { max_age: None, bandwidth: moq_net::bandwidth::Allocator::unlimited(), program: None, + linger: Duration::ZERO, + stitch: false, }; // `m=request` reads a broadcast out; everything else publishes one in. @@ -283,6 +286,11 @@ struct Pending { bandwidth: moq_net::bandwidth::Allocator, /// The programs of a multiplex an ingest publishes. Override with [`Publish::with_program`]. program: Option, + /// How long an egress waits for the same instance to return. Override with + /// [`Subscribe::with_linger`]. + linger: Duration, + /// Whether an egress follows a replacement. Override with [`Subscribe::with_stitch`]. + stitch: bool, } /// What an accepted SRT connection wants: to contribute media ([`Publish`]) or to @@ -443,6 +451,22 @@ impl Subscribe { self.0.peer } + /// Wait up to `linger` 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. + pub fn with_linger(mut self, linger: Duration) -> Self { + self.0.linger = linger; + self + } + + /// Follow another publisher instance replacing the broadcast, as a full program switch on + /// the same SRT connection. Off by default, which ends the stream with + /// [`moq_mux::Error::Replaced`]. + pub fn with_stitch(mut self, stitch: bool) -> Self { + self.0.stitch = stitch; + self + } + /// Accept the subscribe: resolve the broadcast at `path` in `origin`, re-mux /// it to MPEG-TS, and stream it down to the caller until either side ends. /// @@ -453,7 +477,12 @@ impl Subscribe { let path = path.as_path(); let socket = self.0.request.accept(None).await?; tracing::info!(peer = %self.0.peer, %path, "SRT subscribe accepted"); - serve_subscribe(origin, path.as_str(), socket, self.0.latency).await + let options = crate::ts::Options { + latency: self.0.latency, + linger: self.0.linger, + stitch: self.0.stitch, + }; + serve_subscribe(origin, path.as_str(), socket, options).await } /// Reject the subscribe with a verdict the client can distinguish on the wire. @@ -514,7 +543,7 @@ pub(crate) async fn serve_subscribe( origin: &origin::Consumer, path: &str, mut socket: SrtSocket, - latency: Duration, + options: crate::ts::Options, ) -> Result<()> { // Resolve the broadcast, but watch the socket while we wait: `routed` // parks forever for a stream that is never published, and nothing else polls the @@ -526,7 +555,7 @@ pub(crate) async fn serve_subscribe( tracing::debug!(%path, "SRT subscribe closed before its broadcast was available"); return Ok(()); } - subscriber = crate::ts::Subscriber::new(origin, path, latency) => subscriber?, + subscriber = crate::ts::Subscriber::new(origin, path, options) => subscriber?, }; let Some(mut subscriber) = subscriber else { @@ -535,9 +564,22 @@ pub(crate) async fn serve_subscribe( }; let mut egress = Egress::default(); - while let Some(frame) = subscriber.next().await? { - // Sample the muxer's generation alongside the frame it describes, so a - // publisher rewind re-anchors the pacing rather than mapping the new timeline + loop { + // Keep watching the socket: a linger can hold `next` for as long as it is set, and + // nothing is sent meanwhile, so a caller who hangs up would otherwise hold the task. + let next = tokio::select! { + biased; + _ = wait_closed(&mut socket) => { + tracing::debug!(%path, "SRT subscribe closed by the caller"); + return Ok(()); + } + next = subscriber.next() => next?, + }; + let Some(frame) = next else { + break; + }; + // Sample the muxer's generation alongside the frame it describes, so a new + // program clock re-anchors the pacing rather than mapping the new timeline // through the old anchor. for chunk in egress.push(&frame, subscriber.discontinuity(), Instant::now()) { socket.send(chunk).await?; @@ -567,7 +609,7 @@ fn clamp_to_floor(send_at: Instant, floor: &mut Option) -> Instant { /// Resolve once the SRT caller hangs up (a clean close or an error), draining and /// ignoring any unexpected inbound packets. A subscribe caller normally sends -/// nothing, so this is purely a disconnect signal to race against the announce wait. +/// nothing, so this is purely a disconnect signal to race against waiting on the broadcast. async fn wait_closed(socket: &mut SrtSocket) { use futures::TryStreamExt; while let Ok(Some(_)) = socket.try_next().await {} @@ -787,13 +829,14 @@ mod tests { ); } - /// Regression for the SRT half of #2833: the muxer restarts its program clock on a - /// publisher rewind, so an egress that keeps its old anchor maps the whole rewound - /// span into the past, where the first-packet floor collapses it onto one instant. - /// The receiver then sees the program stop until the media climbs back to where it - /// left off, then the accumulated backlog arrive at once. + /// Regression for the SRT half of #2833: a new program clock can start earlier than the + /// old one (a switch to another publisher instance starts its timeline over), so an + /// egress that keeps its old anchor maps the whole new span into the past, where the + /// first-packet floor collapses it onto one instant. The receiver then sees the program + /// stop until the media climbs back to where it left off, then the accumulated backlog + /// arrive at once. #[test] - fn a_rewind_re_anchors_the_pacing() { + fn a_new_clock_re_anchors_the_pacing() { const SLOT: Duration = Duration::from_millis(25); let start = Instant::now(); let mut egress = Egress::default(); @@ -804,10 +847,10 @@ mod tests { assert_eq!(first[0].0, start, "the first payload anchors the connection"); assert_eq!(second[0].0, start + SLOT); - // The publisher rewinds to the top of its source. + // The program switches to an instance whose timeline starts at zero. let mut now = start + 2 * SLOT; - let rewound = egress.push(&frame(0, &[2; SRT_PAYLOAD]), 1, now); - assert_eq!(rewound[0].0, now, "the new generation becomes the live edge"); + let switched = egress.push(&frame(0, &[2; SRT_PAYLOAD]), 1, now); + assert_eq!(switched[0].0, now, "the new generation becomes the live edge"); // The grid slots that follow pace off the new anchor. Against the old one every // one of them is ten minutes in the past and clamps to `start`, so the receiver @@ -821,16 +864,16 @@ mod tests { /// The old generation's buffered tail goes out under the instant it was paced at. /// One SRT message carries one TSBPD timestamp, and the chunker only splits when - /// the instant changes, so the flush has to be explicit: a rewind paced within the + /// the instant changes, so the flush has to be explicit: a new clock paced within the /// same clock tick would otherwise fold media that already played into the new /// program's first payload. #[test] - fn a_rewind_flushes_the_partial_chunk_first() { + fn a_new_clock_flushes_the_partial_chunk_first() { const SLOT: Duration = Duration::from_millis(25); let start = Instant::now(); let mut egress = Egress::default(); - // A partial chunk buffered under the first generation, then a rewind paced at + // A partial chunk buffered under the first generation, then a new clock paced at // the very same instant. assert!(egress.push(&frame(600_000_000, &[1; 188]), 0, start).is_empty()); let chunks = egress.push(&frame(0, &[2; SRT_PAYLOAD]), 1, start); @@ -838,7 +881,7 @@ mod tests { assert_eq!(chunks[0].1.as_ref(), [1u8; 188].as_slice()); assert_eq!(chunks[1].1.as_ref(), [2u8; SRT_PAYLOAD].as_slice()); - // And a tail paced earlier keeps that earlier instant rather than the rewind's. + // And a tail paced earlier keeps that earlier instant rather than the new clock's. assert!(egress.push(&frame(25_000, &[3; 188]), 1, start + SLOT).is_empty()); let chunks = egress.push(&frame(0, &[4; SRT_PAYLOAD]), 2, start + 2 * SLOT); assert_eq!(chunks[0].0, start + SLOT, "the old tail keeps its own instant"); @@ -862,14 +905,14 @@ mod tests { use moq_net::Timestamp; // Short enough to keep the test quick, long enough that the release instants - // either side of the rewind are separated by more than clock noise. + // either side of the marker are separated by more than clock noise. const LATENCY: Duration = Duration::from_millis(300); - // Ten minutes in, the span from the controlled-rewind evidence on #2833. + // Ten minutes in, the span from the evidence on #2833. const OFFSET: u64 = 600_000_000; let (origin, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); tokio::spawn(moq_net::time::run(driver)); - let mut broadcast = origin.create_broadcast("rewind").unwrap(); + let mut broadcast = origin.create_broadcast("marker").unwrap(); broadcast.announce(moq_net::origin::Route::default()).unwrap(); let mut catalog = moq_mux::catalog::Producer::new( &mut broadcast, @@ -911,7 +954,7 @@ mod tests { let addr = server.local_addr(); let caller = tokio::spawn(async move { SrtSocket::builder() - .call(addr, Some("#!::r=rewind,m=request")) + .call(addr, Some("#!::r=marker,m=request")) .await .unwrap() }); @@ -920,7 +963,7 @@ mod tests { panic!("m=request must create a subscribe request"); }; let consumer = origin.consume(); - let egress = tokio::spawn(async move { subscribe.accept(&consumer, "rewind").await }); + let egress = tokio::spawn(async move { subscribe.accept(&consumer, "marker").await }); let mut receiver = caller.await.unwrap(); // Three seconds of program, then wait until the receiver is actually playing it @@ -968,6 +1011,215 @@ mod tests { egress.abort(); } + /// One publisher instance of `live`: an announced broadcast under its own epoch with one + /// Legacy AAC track, for the egress tests. + struct Instance { + _broadcast: moq_net::broadcast::Producer, + catalog: moq_mux::catalog::Producer, + audio: moq_mux::container::Producer, + } + + impl Instance { + fn publish(origin: &moq_net::origin::Producer, track: &str) -> Self { + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + let mut broadcast = origin.publish("live", route).unwrap(); + let mut catalog = moq_mux::catalog::Producer::new( + &mut broadcast, + moq_mux::catalog::Config::default() + .with_catalog(moq_mux::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let audio = broadcast + .create_track(track, hang::container::track_info(hang::catalog::PRIORITY.audio)) + .unwrap(); + let mut config = hang::catalog::AudioConfig::new(hang::catalog::AAC { profile: 2 }, 48_000, 2); + config.container = hang::catalog::Container::Legacy; + catalog + .modify() + .unwrap() + .audio + .renditions + .insert(track.to_string(), config); + let audio = moq_mux::container::Producer::new( + audio, + moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Audio), + ); + Self { + _broadcast: broadcast, + catalog, + audio, + } + } + + /// Write `count` 100ms frames from `offset` microseconds, in one-second groups. + fn write(&mut self, count: u64, offset: u64) { + for i in 0..count { + self.audio + .write(Frame { + timestamp: moq_net::Timestamp::from_micros(offset + i * 100_000).unwrap(), + duration: None, + payload: bytes::Bytes::from_iter((0..180u16).map(|b| (b ^ i as u16) as u8)), + keyframe: i % 10 == 0, + }) + .unwrap(); + if i % 10 == 9 { + self.audio.cut(None).unwrap(); + } + } + } + + /// End the broadcast cleanly. + fn finish(mut self) { + self.audio.finish().unwrap(); + self.catalog.finish().unwrap(); + } + } + + /// Accept one `m=request` caller for `live` with `linger` and `stitch`, returning the + /// receiving socket and the egress task. + async fn egress( + origin: &moq_net::origin::Producer, + linger: Duration, + stitch: bool, + ) -> (SrtSocket, tokio::task::JoinHandle>) { + let mut server = Server::bind("127.0.0.1:0".parse().unwrap(), Duration::from_millis(300)) + .await + .unwrap(); + let addr = server.local_addr(); + let caller = tokio::spawn(async move { + SrtSocket::builder() + .call(addr, Some("#!::r=live,m=request")) + .await + .unwrap() + }); + let Request::Subscribe(subscribe) = server.accept().await.expect("an SRT request") else { + panic!("m=request must create a subscribe request"); + }; + let consumer = origin.consume(); + let task = tokio::spawn(async move { + let _server = server; + subscribe + .with_linger(linger) + .with_stitch(stitch) + .accept(&consumer, "live") + .await + }); + (caller.await.unwrap(), task) + } + + /// The `version_number` of each PAT section a payload starts. + fn pat_versions(payload: &[u8]) -> Vec { + payload + .as_chunks::<188>() + .0 + .iter() + .filter(|p| (u16::from(p[1] & 0x1f) << 8 | u16::from(p[2])) == 0 && p[1] & 0x40 != 0) + .map(|p| { + let start = 4 + if p[3] & 0x20 != 0 { 1 + usize::from(p[4]) } else { 0 }; + let section = start + 1 + usize::from(p[start]); + (p[section + 5] >> 1) & 0x1f + }) + .collect() + } + + /// With stitch, another instance replacing the broadcast switches the program on the same + /// SRT connection: the caller keeps receiving, now under a new PAT version. + #[tokio::test] + async fn a_stitched_replacement_keeps_the_srt_connection() { + let (origin, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); + tokio::spawn(moq_net::time::run(driver)); + let mut old = Instance::publish(&origin, "old.aac"); + let (mut receiver, egress) = egress(&origin, Duration::ZERO, true).await; + + old.write(30, 600_000_000); + let first = tokio::time::timeout(Duration::from_secs(10), receiver.next()) + .await + .expect("the first payload never arrived") + .expect("the SRT egress closed early") + .unwrap(); + assert_eq!(pat_versions(&first.1), [0], "the old program opens the connection"); + + // The replacement starts its own timeline over while the old instance stays up. + let mut new = Instance::publish(&origin, "new.aac"); + new.write(30, 0); + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let (_, payload) = receiver + .next() + .await + .expect("the SRT egress closed at the switch") + .unwrap(); + if pat_versions(&payload).contains(&1) { + break; + } + } + }) + .await + .expect("the switched program never arrived"); + + drop((old, new, receiver)); + egress.abort(); + } + + /// An end that nothing replaces closes the SRT stream once the linger runs out, so a + /// finished broadcast never leaves a caller connected indefinitely. + #[tokio::test] + async fn an_unreplaced_end_closes_the_srt_stream_at_the_linger() { + const LINGER: Duration = Duration::from_secs(1); + let (origin, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); + tokio::spawn(moq_net::time::run(driver)); + let mut live = Instance::publish(&origin, "a.aac"); + let (mut receiver, egress) = egress(&origin, LINGER, false).await; + + live.write(5, 0); + tokio::time::timeout(Duration::from_secs(10), receiver.next()) + .await + .expect("the first payload never arrived") + .expect("the SRT egress closed early") + .unwrap(); + let ended = tokio::time::Instant::now(); + live.finish(); + + tokio::time::timeout(Duration::from_secs(10), async { + while let Some(payload) = receiver.next().await { + payload.unwrap(); + } + }) + .await + .expect("the SRT stream stayed open past the linger"); + assert!(ended.elapsed() >= LINGER, "closed after only {:?}", ended.elapsed()); + egress.await.unwrap().unwrap(); + } + + /// A caller hanging up while the egress lingers ends it then, rather than holding the + /// task and its socket until the linger runs out. + #[tokio::test] + async fn a_caller_hanging_up_during_the_linger_ends_the_egress() { + let (origin, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); + tokio::spawn(moq_net::time::run(driver)); + let mut live = Instance::publish(&origin, "a.aac"); + let (mut receiver, egress) = egress(&origin, Duration::from_secs(600), false).await; + + live.write(5, 0); + tokio::time::timeout(Duration::from_secs(10), receiver.next()) + .await + .expect("the first payload never arrived") + .expect("the SRT egress closed early") + .unwrap(); + live.finish(); + // Take everything the broadcast sent, so the egress is lingering when the caller goes. + while let Ok(Some(payload)) = tokio::time::timeout(Duration::from_secs(2), receiver.next()).await { + payload.unwrap(); + } + receiver.close().await.unwrap(); + + tokio::time::timeout(Duration::from_secs(10), egress) + .await + .expect("the egress outlived its caller") + .unwrap() + .unwrap(); + } + /// Whether a payload carries a TS packet whose adaptation field sets /// `discontinuity_indicator`: the muxer flags the new generation's leading clock /// packet with it and nothing else. diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index 0ec6726721..b3fdf755bb 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -160,20 +160,35 @@ impl Publisher { } } +/// How an egress treats its broadcast ending or being replaced. +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct Options { + /// The muxer's jitter-buffer delay ([`Subscriber::new`]). + pub latency: Duration, + /// How long to wait for the same publisher instance to come back once its broadcast ends. + pub linger: Duration, + /// Follow another instance replacing the broadcast, as a full program switch. + pub stitch: bool, +} + /// Muxes a single MoQ broadcast back into an MPEG-TS byte stream for egress. /// /// The mirror of [`Publisher`]: where that demuxes SRT-carried TS into the /// origin, this consumes a broadcast from the origin and re-muxes it to TS so an /// SRT caller can play it. Pull frames with [`next`](Self::next); each carries /// the TS bytes plus the media timestamp used to pace delivery. +/// +/// The path's announcements drive it through [`ts::Follower`], as they drive a player: the same +/// publisher instance returning within the linger continues the stream, and a replacement either +/// ends it with [`moq_mux::Error::Replaced`] or, with `stitch`, switches the program. pub struct Subscriber { - export: ts::Export, + follower: ts::Follower, } impl Subscriber { /// Resolve the broadcast at `path` in the origin and prepare to mux it to TS. /// - /// `latency` is the muxer's jitter-buffer delay: each frame is muxed that long after + /// `options.latency` is the muxer's jitter-buffer delay: each frame is muxed that long after /// its decode time, a frame arriving later is dropped, and a stalled group is /// skipped after half of it. We reuse the locally configured SRT receive latency for /// it, the same budget an SRT hop gives a packet. It's the configured value, @@ -184,36 +199,39 @@ impl Subscriber { /// Returns `Ok(None)` if the broadcast can never be served (path outside the /// consumer's scope, or the origin closed). Otherwise waits for the broadcast /// to be announced, so a caller may connect before the publisher does. - pub async fn new(origin: &origin::Consumer, path: &str, latency: Duration) -> Result> { - // Confirm the broadcast is in scope and wait for it to be announced (out-of-scope / - // origin-closed -> `None`). The export re-resolves it (and any referenced sibling - // broadcast, via the catalog `broadcast` field) through the origin. + pub(crate) async fn new(origin: &origin::Consumer, path: &str, options: Options) -> Result> { if origin.routed(path).await.is_none() { return Ok(None); } + // The export resolves the broadcast (and any referenced sibling broadcast, via the + // catalog `broadcast` field) through the origin, joining the one just announced. let source = moq_mux::Source::new(origin.consume(), path); let export = ts::Export::with_ts(source, moq_mux::catalog::CatalogFormat::Hang) .await? - .with_delay(latency); - Ok(Some(Self { export })) + .with_delay(options.latency); + let follower = ts::Follower::new(export)? + .with_linger(options.linger) + .with_stitch(options.stitch); + Ok(Some(Self { follower })) } /// Pull the next muxed frame (TS bytes + media timestamp), or `None` once the - /// broadcast ends. + /// broadcast ends with nothing to follow. Dropping the future before it resolves loses + /// nothing. pub async fn next(&mut self) -> Result> { - Ok(self.export.next().await?) + Ok(self.follower.next().await?) } /// The muxer's generation counter for the frame [`next`](Self::next) just - /// returned, which increments each time the publisher rewinds its timeline. + /// returned, which increments each time the program clock restarts: a marker that + /// breaks the publisher's timeline, or a switch to another instance. /// - /// The muxer restarts its program clock across that boundary, so a caller - /// pacing on the media timestamps has to drop its own anchor with it: sample - /// this after every frame and re-anchor when it changes, or the rewound span - /// maps into the past and the whole new generation collapses onto one instant. + /// A caller pacing on the media timestamps has to drop its own anchor with it: sample + /// this after every frame and re-anchor when it changes, or the new clock maps + /// through the old anchor and the whole new generation collapses onto one instant. pub fn discontinuity(&self) -> u64 { - self.export.discontinuity() + self.follower.export().discontinuity() } } @@ -619,7 +637,7 @@ mod tests { let expected_cue = cue.payload; assert_eq!(expected_cue[0], 0xFC, "a verbatim splice_info_section (table_id 0xFC)"); - let mut subscriber = Subscriber::new(&origin.consume(), "ingest", Duration::ZERO) + let mut subscriber = Subscriber::new(&origin.consume(), "ingest", Options::default()) .await .unwrap() .expect("the ingest broadcast is available for SRT egress");