diff --git a/doc/bin/hls.md b/doc/bin/hls.md index c1b86864fa..85b0688e1d 100644 --- a/doc/bin/hls.md +++ b/doc/bin/hls.md @@ -59,20 +59,38 @@ from the replayed timelines alone, and a segment GETs only its rendition's stored objects, so switching renditions never downloads both. An inline-parameter-set codec with no catalog `description` is the exception: the first playlist render GETs one keyframe group to build the init segment, -then caches it. Out-of-band configs need no media GET. When the catalog's -`archive` entry names a `store` and no `replay` path, its spans are durable on -this broadcast, so the playlists list the whole retained timeline and only the -recording's own retention trims them; DASH `timeShiftBufferDepth` is the listed -span. The playlist ends with `EXT-X-ENDLIST` only once the reader's caller +then caches it. Out-of-band configs need no media GET. A recording's playlists +list the same capped window as a live broadcast. A library embedder can set +`export::Config::history` to list past it instead: when the catalog's `archive` +entry names a `store` and no `replay` path, its spans are durable on this +broadcast, so only the recording's own retention trims the playlists and DASH +`timeShiftBufferDepth` is the listed span. The listing starts at the records +the timeline restates when the exporter joins (at most 256 from a `moq-mux` +publisher). The playlist ends with `EXT-X-ENDLIST` only once the reader's caller declares the recording finished; the store holds no completion marker. +A viewer joining a days-old broadcast costs what the window lists, not the +broadcast's age: the exporter reads only the records its timeline restates on +join, never the whole history. The window lists at most 256 +segments however short they are, so every edge shows the same +`EXT-X-MEDIA-SEQUENCE`. Below that cap the window never evicts its newest +segment that starts on a video sync point, so a GOP longer than `--window` +stretches the window to that one GOP rather than leave nothing a player can +start at. A rendition with no media for a span lists that span as +a duration-preserving `EXT-X-GAP` and goes on listing after it. The master +playlist advertises a video rendition only once a listed segment starts at a +group start that is a sync point, so an early master may list audio alone, and +answers 404 while no rendition can start. Once advertised, a rendition stays in +the master for the rest of the publisher run, even if a GOP longer than 256 +records evicts its sync segment. + The init URL carries a hash of its bytes, so a reconfigured rendition gets a new one. An embedder of the library can also label the publisher's run with `Broadcaster::set_generation`. Every segment URL then carries it (`seg/{generation}.{reference}.{segment}.m4s`), since a restarted publisher reuses segment numbers for different media. -`--window` sets the live playlist duration (default 16 s) and caps segment +`--window` sets the playlist duration (default 16 s) and caps segment `Cache-Control: max-age` for every broadcast, `--listen-tls-cert`/`--listen-tls-key` or `--listen-tls-generate` serve HTTPS, and `--cors-origin` opens it to browsers. diff --git a/quest/m1/README.md b/quest/m1/README.md index 76699d10e7..6c4eb3e3f8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -82,7 +82,6 @@ blocks. The quests that gated m0 lines moved under them. - [JS IETF datagrams](/quest/m1/js-ietf-datagram.md) - `@moq/net` sends and receives datagram groups over moq-transport, like Rust - [Datagrams are live-only](/quest/m1/datagram-unfetchable.md) - no FETCH, replay to a new subscriber, or cache fill ever returns a datagram group - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface -- [Bounded HLS playlists](/quest/m1/hls-bounded.md) - `moq-hls` renders a capped sliding window for a fresh viewer, with a stable media sequence, gap slots, and no video before a sync point - [JS request window](/quest/m1/js-request-window.md) - @moq/net grants MAX_REQUEST_ID as requests close, so a peer never stalls after ~21k requests - [Relay session limits](/quest/m1/relay-session-limits.md) - moq-relay sets per-session request limits, tighter for clients than peers, and the bindings name a refused request - [Churn with held subscriptions](/quest/m1/session-churn-held.md) - opening and closing a request costs the same with 1 or 1,024 held subscriptions diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index feb8ad71e6..d5aad63e07 100644 --- a/quest/m1/archive/README.md +++ b/quest/m1/archive/README.md @@ -49,9 +49,12 @@ Every track has its own timeline (`rs/moq-mux/src/timeline.rs`): playlists read only the timelines, and a segment GETs only its rendition's objects. The caller supplies the catalog. - A catalog `archive` entry with a `store` and no `replay` path declares its - spans durable on that broadcast, so the exporter lists the whole retained - timeline and only its pops trim it (`durable` in - `rs/moq-hls/src/export/mod.rs`). + spans durable on that broadcast. The exporter still lists a capped window + for it; only the explicit `export::Config::history` mode lists the whole + retained timeline and lets only its pops trim it (`durable` in + `rs/moq-hls/src/export/mod.rs`). Either way the listing starts at the + records the timeline restates on join (at most 256 from `moq-mux`), so a + longer recording does not list from its start yet. `rs/moq-archive` stores the versioned objects on any `object_store::ObjectStore`. `moq_archive::Writer` (`rs/moq-archive/src/writer.rs`) cuts each enrolled track @@ -167,6 +170,5 @@ owned by that prerequisite, not duplicated in archive storage. ## Related -- [Bounded HLS playlists](/quest/m1/hls-bounded.md) - the live renderer rules a replayed archive inherits, planned outside this line - [Catalog track identity](/quest/m2/catalog-tracks.md) - explore immutable definitions or explicit version binding independently of archives - [e2ee](/quest/m1/e2ee/README.md) - protected broadcasts are excluded initially diff --git a/quest/m1/archive/dvr.md b/quest/m1/archive/dvr.md index b0885ab506..367c01531b 100644 --- a/quest/m1/archive/dvr.md +++ b/quest/m1/archive/dvr.md @@ -42,7 +42,9 @@ catalog](/quest/m1/archive/replay-catalog.md) now plans. During live, rewind needs no handover: a recorded broadcast's live timeline is durable, and every group it lists is promised available, so a seek past the live window FETCHes old groups through the normal miss chain down to the -archive. `moq-hls` needs no special path. `moq-hls` gains +archive. A live HLS playlist stays capped to its window even for a durable +timeline (decided 10-05), so an HLS seek past it needs the explicit +`moq_hls::export::Config::history` listing. `moq-hls` gains no linger (the `hls-linger` quest was dropped, because an unannounced broadcast can't be FETCHed and a linger would only serve the cache). Test the handover: a live HLS session keeps its playlist URIs and media sequence diff --git a/quest/m1/archive/replay-catalog.md b/quest/m1/archive/replay-catalog.md index 51faba1ca7..5e1e62e139 100644 --- a/quest/m1/archive/replay-catalog.md +++ b/quest/m1/archive/replay-catalog.md @@ -6,8 +6,8 @@ A replayed recording plays as VOD HLS from the stock server: `moq export archive`, then `moq import archive`, then `moq export hls` lists the whole recording with no embedder supplying a catalog. The replay broadcast publishes its recorded catalog live, stamped with the recording's `store` and `version`, -so the exporter treats its timeline as durable and lists past `--window`. A -`--follow` replay grows like an event playlist. +so an exporter in history mode treats its timeline as durable and lists past +`--window`. A `--follow` replay grows like an event playlist. Choosing which catalog applies to which media group stays with [Catalog track identity](/quest/m2/catalog-tracks.md). @@ -36,6 +36,14 @@ this by hand-building a catalog. timelines, not the recording's, so replace it with the timelines the reader replays rather than trusting the recorded ones. +A live playlist stays capped even for a durable timeline; listing past the +window is the explicit `moq_hls::export::Config::history` mode, which `moq +export hls` does not expose yet. Give it a flag here. History mode lists from +the records the timeline restates when the exporter joins (at most 256 from a +`moq-mux` publisher), so "the whole recording" also needs the exporter to +read a longer recording's stored history rather than only its live +restatement. + Add a CI test that records a broadcast longer than the default window, replays it, and asserts the stock exporter lists segment 0 with a durable `timeShiftBufferDepth`. Consider moving the HLS archive tests onto the diff --git a/quest/m1/hls-bounded.md b/quest/m1/hls-bounded.md deleted file mode 100644 index a858c01c0f..0000000000 --- a/quest/m1/hls-bounded.md +++ /dev/null @@ -1,87 +0,0 @@ -# [M] Bounded HLS playlists - -## Goal - -A fresh viewer joins a days-old broadcast, live or archived, and `moq-hls` -renders its playlists with work bounded by the advertised window, not by the -broadcast's age. The media playlist lists a sliding window whose -`EXT-X-MEDIA-SEQUENCE` every edge and every reload agree on. A span with no -media in one rendition keeps its slot as a duration-preserving `EXT-X-GAP` -without holding back later segments or sibling renditions. A video rendition -is not advertised until its window holds a segment that starts at a group -start (a sync point), so a stock player is never sent to video it cannot -start decoding. - -This is the generic renderer. Managed edges (moq.pro) only wire it to their -projects, routes, and auth. - -## Plan - -Decided 2026-10-05, from moq.pro's -[bounded live playlist](https://github.com/moq-dev/moq.pro/blob/main/quest/m2/archive/playable.md): -the windowed sequence, gap slots, and sync-point gating are renderer rules -every `moq-hls` user needs, so they live here and moq.pro keeps only the edge -wiring. Rejected: keeping them in moq.pro. Also decided 2026-10-05: this -quest lives at the m1 root on `main`, outside the archive line, since live -timelines need it first. - -The offline archive HLS that `quest/m1/archive/hls.md` used to describe -landed on `main` with the archive line (#4034: #4115, #4155, #4169). So did -the per-track gap behaviour from per-track timelines (#4280, `spans.rs`), -which derives segments at the edge from group timestamps and emits -`EXT-X-GAP` for a rendition with no group start in a span. The rules here -apply to that shape. - -Start by auditing `rs/moq-hls/src/export` against each rule; much exists -(`segments.rs` numbers aligned segments and carries gap rows, `playlist.rs` -writes `EXT-X-GAP`). Fix only what a test shows missing: - -- **Bounded join.** Rendering for a new viewer reads at most the window's - records from the timeline (`moq_json::window`; a new group repeats at most - `CHECKPOINT_RECORDS = 256` recent records, in `moq-mux`'s `timeline.rs`), - never the whole history, and never GETs or FETCHes media. A live publisher - that never pops its timeline must not make a join cost grow with broadcast - age. The window is bounded by duration, so very short segments still mean - many records per window: decide while building whether a record cap also - applies, and test dense input against it. -- **Capped window, decided 2026-10-05.** A durable (store-backed) timeline - still advertises a capped sliding window: a live playlist lists a bounded - window even when the store retains more. A full VOD or EVENT listing is a - separate, explicit replay mode. This reverses #4155, now on `main`: a - durable timeline (`durable()` in `export/mod.rs`) lists everything it - retains, past `Config::window`, and - `a_durable_timeline_lists_past_the_window` pins it. This quest owns the - reconciliation (decided 2026-10-05): the durable listing becomes this - capped window, the full listing becomes the explicit replay mode, and that - test and the `Config::window` docs change with it. -- **Stable sequence.** `EXT-X-MEDIA-SEQUENCE` is the window's first segment - number, derived from the timeline, so two edges and a reload after a pop - agree. -- **Gaps.** A span with no media in one track becomes a gap slot in that - rendition only; siblings keep listing, and a rendition switch after a gap - lands on the next real segment. A failed rendition timeline is not a gap: - per #4552 its playlist ends at its last covered segment. -- **Sync-point gating.** The multivariant playlist omits a video rendition - until its window holds a segment starting at a group start that is a sync - point. Phrase the check by group start and sync point, not by keyframe, so - it stays right for intra-refresh streams and agrees with - [Export sync flags](/quest/m2/intra-refresh/export-sync-flags.md). A - catalog that arrives first may expose audio only, or answer unavailable. - -Prove: a fresh join after simulated days of publishing reads a bounded number -of records (count them); a durable timeline holding more than the window -still lists only the window; catalog arrival before the first group start; -gaps in one track; switching after a gap; healthy siblings keep their -listing; two exporters over one timeline render identical sequence numbers. -Tests mock time. - -moq.pro pins `release`, so it adopts this through the next release that -carries it. - -Public API: none expected. Wire: none. - -## Related - -- [moq.pro: bounded live playlist](https://github.com/moq-dev/moq.pro/blob/main/quest/m2/archive/playable.md) - the edge wiring that adopts this renderer -- [Archive](/quest/m1/archive/README.md) - a replayed archive inherits these rules -- [Export sync flags](/quest/m2/intra-refresh/export-sync-flags.md) - which group starts the HLS export marks as sync points diff --git a/rs/moq-cli/src/hls.rs b/rs/moq-cli/src/hls.rs index 9d754ef295..f0b5e8facb 100644 --- a/rs/moq-cli/src/hls.rs +++ b/rs/moq-cli/src/hls.rs @@ -32,9 +32,9 @@ pub struct ExportArgs { #[usage(flatten)] pub tls: moq_tokio::tls::Listen, - /// Minimum media listed in each rendition's playlist window. Keep it within the - /// relay's group-cache retention, since segments are fetched from there on request. - /// A timeline durable in a catalog-named store lists everything it retains instead. + /// Minimum media listed in each rendition's playlist window, at most 256 segments. + /// Keep it within the relay's group-cache retention, since segments are fetched from + /// there on request. #[usage(long, default = "16s")] pub window: crate::duration::Duration, diff --git a/rs/moq-hls/src/export/archive_tests.rs b/rs/moq-hls/src/export/archive_tests.rs index 2afa12bf1c..6a0db647df 100644 --- a/rs/moq-hls/src/export/archive_tests.rs +++ b/rs/moq-hls/src/export/archive_tests.rs @@ -290,6 +290,10 @@ struct Replay { impl Replay { async fn open(recording: &Recording, cache: u64, archive: hang::catalog::Archive) -> Self { + Self::open_with(recording, cache, archive, Config::default()).await + } + + async fn open_with(recording: &Recording, cache: u64, archive: hang::catalog::Archive, config: Config) -> Self { let (origin, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::default()); tokio::spawn(moq_net::time::run(driver)); let broadcast = origin.create_broadcast("rec").unwrap(); @@ -302,8 +306,8 @@ impl Replay { let mut catalog = moq_json::snapshot::Producer::new(track, json); catalog.update(&self::catalog(archive)).unwrap(); - let config = reader::Config::new(recording.timelines()).with_cache(cache); - let reader = moq_archive::Reader::open(recording.store.clone(), &broadcast, config) + let serving = reader::Config::new(recording.timelines()).with_cache(cache); + let reader = moq_archive::Reader::open(recording.store.clone(), &broadcast, serving) .await .unwrap(); tokio::spawn(reader.serve()); @@ -313,7 +317,7 @@ impl Replay { } let source = moq_mux::Source::new(origin.consume(), "rec"); - let broadcaster = Broadcaster::new(source, Config::default()).await.unwrap(); + let broadcaster = Broadcaster::new(source, config).await.unwrap(); tokio::time::timeout(Duration::from_secs(5), broadcaster.ready()) .await .expect("a rendition becomes playable"); @@ -564,13 +568,33 @@ async fn a_growing_recording_ends_only_on_caller_finality() { assert!(playlist.contains("seg/29b1cd67.3.m4s\n#EXT-X-ENDLIST\n"), "{playlist}"); } -/// A durable timeline lists the whole recording past the default 16s window, and DASH offers -/// the whole listed span. A live-style entry, or a `replay` path that moves the durable ranges -/// to another broadcast, keeps the window. +/// A durable timeline still lists only the default 16s window: a live playlist stays bounded +/// however much the store retains. #[tokio::test] -async fn a_durable_timeline_lists_past_the_window() { +async fn a_durable_timeline_lists_the_window() { let recording = segments(12).await; let replay = Replay::open(&recording, 64 * 1024 * 1024, durable()).await; + let playlist = replay + .playlist_until(Kind::Video, "1080p", |playlist| { + playlist.contains("seg/29b1cd67.11.m4s\n") + }) + .await; + // Eight 2s segments cover the window. + assert!(playlist.contains("#EXT-X-MEDIA-SEQUENCE:4\n"), "{playlist}"); + assert!(!playlist.contains("seg/29b1cd67.3.m4s\n"), "{playlist}"); +} + +/// History mode lists a durable timeline's whole recording, and DASH offers the whole listed +/// span. A live-style entry, or a `replay` path that moves the durable ranges to another +/// broadcast, keeps the window even then. +#[tokio::test] +async fn history_mode_lists_a_durable_timeline_past_the_window() { + let recording = segments(12).await; + let config = Config { + history: true, + ..Config::default() + }; + let replay = Replay::open_with(&recording, 64 * 1024 * 1024, durable(), config.clone()).await; let playlist = replay .playlist_until(Kind::Video, "1080p", |playlist| { playlist.contains("seg/29b1cd67.11.m4s\n") @@ -588,8 +612,8 @@ async fn a_durable_timeline_lists_past_the_window() { let mut elsewhere = durable(); elsewhere.replay = Some(moq_net::path::RelativeOwned::new("./recording")); for archive in [live(), elsewhere] { - let live = Replay::open(&recording, 64 * 1024 * 1024, archive).await; - let playlist = live + let windowed = Replay::open_with(&recording, 64 * 1024 * 1024, archive, config.clone()).await; + let playlist = windowed .playlist_until(Kind::Video, "1080p", |playlist| { playlist.contains("seg/29b1cd67.11.m4s\n") }) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index aac8e24e0c..c0a07b11d0 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -70,19 +70,35 @@ const CATALOG_RETRY_MAX: Duration = Duration::from_secs(5); #[non_exhaustive] pub struct Config { /// Minimum duration of media listed in each rendition's playlist window. Older timeline - /// records are evicted once the remaining segments still cover this span; keep it within - /// the relay's group-cache retention, since segments are fetched from there on request. + /// records are evicted once the remaining segments still cover this span, except the newest + /// one a video player can start at, so a GOP longer than the window stretches it. Past 256 + /// segments they are evicted however short they are. Keep it within the relay's group-cache retention, since + /// segments are fetched from there on request. It also caps segment `Cache-Control: max-age`. /// - /// A durable timeline (its catalog `archive` entry names a `store` and no `replay` path) - /// lists everything it retains instead, since its own retention already bounds it. The - /// window still caps segment `Cache-Control: max-age` for every broadcast. + /// A durable timeline gets the same window: a live playlist stays bounded however much its + /// store retains. See [`history`](Self::history) to list all of it. pub window: Duration, + + /// List a durable timeline past the window, trimmed only by the timeline's own pops, so a + /// viewer can seek back through a recording. Off by default. + /// + /// A timeline is durable when its catalog `archive` entry names a `store` and no `replay` + /// path, so every range it lists stays FETCHable from this broadcast. Other timelines keep + /// the window, since their segments leave the relay cache. + /// + /// The listing starts at the records the timeline restates when this exporter joins (a + /// `moq-mux` publisher restates at most 256), so it reaches the start of a longer recording + /// only if the exporter followed it from there. For the same reason, edges that joined at + /// different times can list different `EXT-X-MEDIA-SEQUENCE` values: only the window keeps + /// every edge agreeing. + pub history: bool, } impl Default for Config { fn default() -> Self { Self { window: Duration::from_secs(16), + history: false, } } } @@ -123,7 +139,7 @@ impl Broadcaster { }); // The watcher owns its own producer clone; the `Broadcaster`'s `Drop` aborts it so the // standing catalog subscription stops when nobody's serving from this broadcaster. - let watcher = tokio::spawn(watch_catalog(upstream, renditions, timeline_watcher)); + let watcher = tokio::spawn(watch_catalog(upstream, renditions, timeline_watcher, config.history)); *broadcaster.watcher.lock().unwrap() = Some(watcher); Ok(broadcaster) } @@ -157,6 +173,12 @@ impl Broadcaster { self.renditions.ready().await; } + /// Resolve once the master playlist lists at least one rendition. + #[cfg_attr(not(feature = "server"), allow(dead_code))] + pub(crate) async fn advertised(&self) { + self.renditions.advertised().await; + } + /// Version every segment URL with `generation`, a label for the publisher run the /// broadcast carries now, or stop versioning with `None` (the default). /// @@ -188,11 +210,21 @@ impl Broadcaster { /// argument rather than a [`Config`] field because one broadcaster fans out to viewers /// holding different tokens; a credential in `Config` would embed one viewer's in /// another's playlist. + /// + /// Lists only renditions a player can start: a video rendition appears once a listed segment + /// starts at a group start that is a sync point, so an early render may list audio alone, or + /// nothing at all. pub fn master_playlist(&self, query: Option<&str>) -> String { + let (video, audio) = self.variants(query); + master::render(&video, &audio) + } + + /// The variants [`master_playlist`](Self::master_playlist) lists. + pub(crate) fn variants(&self, query: Option<&str>) -> (Vec, Vec) { let mut video = Vec::new(); let mut audio = Vec::new(); for rendition in self.renditions.snapshot() { - if !rendition.is_playable() { + if !rendition.is_advertised() { continue; } match rendition.kind { @@ -211,7 +243,7 @@ impl Broadcaster { }), } } - master::render(&video, &audio) + (video, audio) } /// Render the DASH manifest (MPD) from the current renditions and their views of the @@ -325,6 +357,7 @@ async fn watch_catalog( upstream: Upstream, renditions: renditions::Producer, timeline_watcher: Arc>>>, + history: bool, ) { // The already-resolved handle rather than a fresh lookup through `source`, so the catalog, // the timeline subscription, and the closed check below all refer to the same resolution. A @@ -355,7 +388,7 @@ async fn watch_catalog( match kio::wait(|waiter| consumer.poll_next(waiter)).await { Ok(Some(catalog)) => { // Decide the playlists' retention before any rendition starts following its timeline. - if !unbound && catalog.archive.as_ref().is_some_and(durable) { + if history && !unbound && catalog.archive.as_ref().is_some_and(durable) { unbound = true; renditions.fanout().unbound(); } @@ -417,8 +450,8 @@ fn reference(catalog: &moq_mux::catalog::hang::Catalog) -> Option { /// Whether every range `archive` advertises stays FETCHable from this broadcast until the /// timeline pops it: a store makes the ranges durable, and no `replay` path means this -/// broadcast serves them. The catalog states this, so the playlists follow the timeline's own -/// retention rather than a window sized for relay caches. +/// broadcast serves them. The catalog states this, so [`Config::history`] can follow the +/// timeline's own retention rather than a window sized for relay caches. fn durable(archive: &hang::catalog::Archive) -> bool { archive.store.is_some() && archive.replay.is_none() } @@ -2852,38 +2885,69 @@ mod tests { ) } - /// A `video0` reference and a `video1` rendition over hand-published timelines, reconciled and - /// watched the way `watch_catalog` does. Nothing serves the media, so a segment fetch is a miss. - struct Failing { + /// Renditions over hand-published timelines, reconciled and watched the way `watch_catalog` + /// does. Nothing serves the media, so a segment fetch is a miss. + struct Timelines { broadcast: moq_net::broadcast::Producer, + catalog: moq_mux::catalog::hang::Catalog, renditions: renditions::Producer, watcher: tokio::task::JoinHandle<()>, } - impl Failing { + impl Timelines { + /// A `video0` reference and a `video1` rendition. fn new() -> Self { + Self::with(&["video0", "video1"], &[]) + } + + /// `video` and `audio` renditions, each indexing its own timeline; the first video one is + /// the reference. + fn with(video: &[&str], audio: &[&str]) -> Self { + let mut catalog = moq_mux::catalog::hang::Catalog::default(); + for name in video { + catalog.video.renditions.insert(name.to_string(), video_config()); + } + for name in audio { + let config = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2); + catalog.audio.renditions.insert(name.to_string(), config); + } + let tracks: Vec<&str> = video.iter().chain(audio).copied().collect(); + catalog.archive = Some(archive(&tracks)); let broadcast = moq_net::broadcast::Info::new().produce(); + let (renditions, watcher) = Self::export(&broadcast, &catalog); + Self { + broadcast, + catalog, + renditions, + watcher, + } + } + + fn export( + broadcast: &moq_net::broadcast::Producer, + catalog: &moq_mux::catalog::hang::Catalog, + ) -> (renditions::Producer, tokio::task::JoinHandle<()>) { let upstream = Upstream { source: moq_mux::Source::new(produce_origin().consume(), "live"), broadcast: broadcast.consume(), }; - let mut catalog = moq_mux::catalog::hang::Catalog::default(); - for name in ["video0", "video1"] { - catalog.video.renditions.insert(name.to_string(), video_config()); - } - let archive = archive(&["video0", "video1"]); - catalog.archive = Some(archive.clone()); - let renditions = renditions::Producer::new(Config::default().window); - renditions.sync(&upstream, &catalog); + renditions.sync(&upstream, catalog); let watcher = tokio::spawn(watch_timeline( upstream.broadcast.clone(), - archive, - reference(&catalog).unwrap(), + catalog.archive.clone().unwrap(), + reference(catalog).unwrap(), renditions.fanout(), )); + (renditions, watcher) + } + + /// Another exporter over the same timelines: a second edge, or one a fresh viewer starts. + fn join(&self) -> Self { + let (renditions, watcher) = Self::export(&self.broadcast, &self.catalog); Self { - broadcast, + broadcast: self.broadcast.clone(), + catalog: self.catalog.clone(), renditions, watcher, } @@ -2900,7 +2964,10 @@ mod tests { } fn rendition(&self, name: &str) -> Arc { - self.renditions.get(Kind::Video, name).unwrap() + self.renditions + .get(Kind::Video, name) + .or_else(|| self.renditions.get(Kind::Audio, name)) + .unwrap() } /// Wait until `name` lists segment `last`. @@ -2938,7 +3005,7 @@ mod tests { } } - impl Drop for Failing { + impl Drop for Timelines { fn drop(&mut self) { self.watcher.abort(); } @@ -2958,7 +3025,7 @@ mod tests { // records cover, rather than listing the rest as gaps, while the other renditions go on. #[tokio::test(start_paused = true)] async fn a_rendition_timeline_error_ends_its_playlist() { - let test = Failing::new(); + let test = Timelines::new(); let (_reference, mut video0) = test.publish("video0"); let (track, mut video1) = test.publish("video1"); for sequence in 0..4 { @@ -2988,7 +3055,7 @@ mod tests { // of freezing live, and recording cursors end. #[tokio::test(start_paused = true)] async fn a_reference_timeline_error_ends_every_playlist() { - let test = Failing::new(); + let test = Timelines::new(); let (track, mut video0) = test.publish("video0"); let (_track, mut video1) = test.publish("video1"); for sequence in 0..4 { @@ -3016,7 +3083,7 @@ mod tests { // resolve to, so a listed segment never turns into a gap. #[tokio::test(start_paused = true)] async fn a_rendition_ahead_of_the_reference_keeps_listed_segments() { - let test = Failing::new(); + let test = Timelines::new(); let (_track0, mut video0) = test.publish("video0"); let (_track1, mut video1) = test.publish("video1"); // The reference stalls at 20s while video1 runs on to 80s, far past the 16s window. @@ -3039,7 +3106,7 @@ mod tests { // never reuse one the old reference listed for other content. #[tokio::test(start_paused = true)] async fn a_reference_switch_never_reuses_a_segment_url() { - let mut test = Failing::new(); + let mut test = Timelines::new(); let (_track0, mut video0) = test.publish("video0"); let (_track1, mut video1) = test.publish("video1"); for sequence in 0..4 { @@ -3084,7 +3151,7 @@ mod tests { // A malformed reference timeline fails the same way. #[tokio::test(start_paused = true)] async fn a_malformed_reference_timeline_ends_every_playlist() { - let test = Failing::new(); + let test = Timelines::new(); let (track, mut video0) = test.publish("video0"); video0.push(&gop(0)).unwrap(); test.listed("video0", 0).await; @@ -3099,7 +3166,7 @@ mod tests { // parks. #[tokio::test(start_paused = true)] async fn a_pending_timeline_subscription_ends_with_the_broadcast() { - let test = Failing::new(); + let test = Timelines::new(); // A handler keeps the unpublished timelines' requests waiting rather than refused. let _handler = test.broadcast.dynamic(); tokio::time::sleep(Duration::from_secs(60)).await; @@ -3109,4 +3176,322 @@ mod tests { test.finished("video0").await; test.drain("video0").await; } + + /// Record `sequence` covering group `group` from `pts_ms` for `duration_ms`, starting on a sync + /// point. + fn span(sequence: u64, pts_ms: u64, duration_ms: u64, group: u64) -> hang::timeline::Record { + hang::timeline::Record::new( + sequence, + pts_ms, + duration_ms, + hang::timeline::Position::group(group), + hang::timeline::Position::group(group + 1), + ) + } + + /// The renditions a master playlist would list, by name. + fn advertised(renditions: &renditions::Producer) -> Vec { + renditions + .snapshot() + .iter() + .filter(|rendition| rendition.is_advertised()) + .map(|rendition| rendition.name.clone()) + .collect() + } + + /// The `Push` events a fresh subscriber to `track`'s timeline reads before it is caught up. + async fn fresh_join_reads(test: &Timelines, track: &str) -> usize { + let section = test.catalog.archive.clone().unwrap(); + let mut timeline = moq_mux::timeline::Consumer::<()>::subscribe(&test.broadcast.consume(), §ion, track) + .await + .unwrap(); + let mut pushes = 0; + while let std::task::Poll::Ready(event) = timeline.poll_next(&kio::Waiter::noop()) { + let event = event.unwrap().expect("the timeline is still live"); + pushes += matches!(event, moq_mux::timeline::Event::Push { .. }) as usize; + } + pushes + } + + // A viewer joining a days-old live broadcast costs what the window lists, not the broadcast's + // age: its timeline subscription restates a bounded suffix of a timeline that never pops, and + // its playlist lists the window with the sequence every other edge shows. + #[tokio::test(start_paused = true)] + async fn a_fresh_join_reads_a_bounded_number_of_records() { + let early = Timelines::with(&["video0"], &[]); + let (_track, mut video0) = early.publish("video0"); + // A day of 10s GOPs per round. + const DAY: u64 = 86_400 / 10; + // A checkpoint, plus the edits one timeline group holds before the publisher rolls it. + let bound = segments::MAX_SEGMENTS * (1 + moq_json::window::ProducerConfig::default().op_ratio as usize); + for day in 1..=3 { + for sequence in (day - 1) * DAY..day * DAY { + video0 + .push(&span(sequence, sequence * 10_000, 10_000, sequence)) + .unwrap(); + } + let read = fresh_join_reads(&early, "video0").await; + assert!( + (segments::MAX_SEGMENTS..=bound).contains(&read), + "day {day}: a fresh join read {read} records" + ); + } + + let last = 3 * DAY - 1; + let late = early.join(); + let joined = late.listed("video0", last).await; + let followed = early.listed("video0", last).await; + // Two 10s segments cover the 16s window. + assert_eq!(numbers(&joined), [last - 1, last]); + assert_eq!(joined.media_sequence, last - 1); + assert_eq!(numbers(&followed), numbers(&joined)); + assert_eq!(followed.media_sequence, joined.media_sequence); + } + + // Very short segments: the window is capped at what a fresh join can see, so an edge that + // followed the broadcast all along lists exactly what one joining now does. + #[tokio::test(start_paused = true)] + async fn a_dense_timeline_lists_the_same_window_on_every_edge() { + let early = Timelines::with(&["video0"], &[]); + let (_track, mut video0) = early.publish("video0"); + // 40ms records: the 16s window spans 400 of them. + for sequence in 0..1_000 { + video0.push(&span(sequence, sequence * 40, 40, sequence)).unwrap(); + settle().await; + } + // The publisher rolls its timeline group, so a joiner reads only the checkpoint and on. + video0.flush().unwrap(); + video0.push(&span(1_000, 40_000, 40, 1_000)).unwrap(); + + let late = early.join(); + let joined = late.listed("video0", 1_000).await; + let followed = early.listed("video0", 1_000).await; + assert_eq!(joined.segments.len(), segments::MAX_SEGMENTS); + assert_eq!(numbers(&followed), numbers(&joined)); + assert_eq!(followed.media_sequence, joined.media_sequence); + } + + // Two edges over one timeline agree on the media sequence, and so does a reload after the + // publisher pops records out of the window. + #[tokio::test(start_paused = true)] + async fn edges_and_reloads_agree_on_the_media_sequence() { + let early = Timelines::with(&["video0"], &[]); + let (_track, mut video0) = early.publish("video0"); + for sequence in 0..20 { + video0.push(&gop(sequence)).unwrap(); + settle().await; + } + let late = early.join(); + let joined = late.listed("video0", 19).await; + let followed = early.listed("video0", 19).await; + assert_eq!(joined.media_sequence, 12, "eight 2s segments cover the 16s window"); + assert_eq!(numbers(&followed), numbers(&joined)); + + // A DVR expiry trims into the window. + video0.pop(15).unwrap(); + video0.push(&gop(20)).unwrap(); + let joined = late.listed("video0", 20).await; + let followed = early.listed("video0", 20).await; + let fresh = early.join().listed("video0", 20).await; + for snapshot in [&joined, &followed, &fresh] { + assert_eq!(snapshot.media_sequence, 15); + assert_eq!(numbers(snapshot), [15, 16, 17, 18, 19, 20]); + } + } + + // A video rendition is advertised once a listed segment starts at a group start that is a + // sync point. Audio needs none, so a master rendered early may list audio alone. + #[tokio::test(start_paused = true)] + async fn video_is_advertised_from_its_first_sync_point() { + let test = Timelines::with(&["video0", "video1"], &["audio0"]); + assert!(advertised(&test.renditions).is_empty(), "nothing listed yet"); + + let (_track0, mut video0) = test.publish("video0"); + let (_track1, mut video1) = test.publish("video1"); + let (_track2, mut audio0) = test.publish("audio0"); + for sequence in 0..4 { + // Joined mid-group, then a group start that is not a sync point (an intra-refresh run). + let mut record = gop(sequence); + record.keyframe = false; + if sequence < 2 { + record.start.frame = 5; + } + video0.push(&record).unwrap(); + video1.push(&record).unwrap(); + audio0.push(&gop(sequence)).unwrap(); + } + test.listed("video0", 3).await; + test.listed("audio0", 2).await; + assert_eq!(advertised(&test.renditions), ["audio0"]); + + // The master route long-polls on this: it must wake on the first sync start. + let renditions = test.renditions.clone(); + let video = tokio::spawn(async move { + kio::wait(|waiter| { + let snapshot = renditions.snapshot(); + let video0 = snapshot.iter().find(|rendition| rendition.name == "video0").unwrap(); + video0.poll_advertised(waiter) + }) + .await + }); + tokio::time::sleep(Duration::from_secs(1)).await; + assert!(!video.is_finished(), "video0 has no sync start yet"); + + for sequence in 4..7 { + video0.push(&gop(sequence)).unwrap(); + video1.push(&gop(sequence)).unwrap(); + audio0.push(&gop(sequence)).unwrap(); + } + tokio::time::timeout(Duration::from_secs(5), video) + .await + .expect("the wait wakes on the first sync start") + .unwrap(); + test.listed("video1", 4).await; + assert_eq!(advertised(&test.renditions), ["video0", "video1", "audio0"]); + } + + /// Record `sequence` covering frames `frames` of one long group from `pts_ms`, a sync point + /// only at the group start: how `moq-mux` splits a GOP longer than its record limit. + fn split(sequence: u64, pts_ms: u64, duration_ms: u64, frames: std::ops::Range) -> hang::timeline::Record { + let mut record = span(sequence, pts_ms, duration_ms, 0); + record.start.frame = frames.start; + record.end = record.start; + record.end.frame = frames.end; + record.keyframe = frames.start == 0; + record + } + + // A 30s GOP is split into 10s records with one sync start. The 16s window keeps that record + // rather than evict it, so video stays startable and advertised until the next GOP starts. + #[tokio::test(start_paused = true)] + async fn a_long_gop_keeps_its_sync_start_in_the_window() { + let test = Timelines::with(&["video0"], &["audio0"]); + let (_track0, mut video0) = test.publish("video0"); + let (_track1, mut audio0) = test.publish("audio0"); + for sequence in 0..3 { + let frames = sequence * 300..(sequence + 1) * 300; + video0 + .push(&split(sequence, sequence * 10_000, 10_000, frames)) + .unwrap(); + audio0 + .push(&span(sequence, sequence * 10_000, 10_000, sequence)) + .unwrap(); + } + let window = test.listed("video0", 2).await; + assert_eq!(numbers(&window), [0, 1, 2], "the sync start stays listed"); + let joined = test.join().listed("video0", 2).await; + assert_eq!(numbers(&joined), [0, 1, 2], "a fresh edge lists the same window"); + assert_eq!(advertised(&test.renditions), ["video0", "audio0"]); + + video0.push(&span(3, 30_000, 10_000, 1)).unwrap(); + audio0.push(&span(3, 30_000, 10_000, 3)).unwrap(); + let window = test.listed("video0", 3).await; + assert_eq!(numbers(&window), [2, 3], "the next sync start releases the old GOP"); + assert_eq!(advertised(&test.renditions), ["video0", "audio0"]); + } + + // A GOP past the 256-record cap loses its sync start to the cap. The rendition stays + // advertised for the rest of the run rather than vanish from a master players already loaded. + #[tokio::test(start_paused = true)] + async fn a_gop_past_the_cap_stays_advertised() { + let test = Timelines::with(&["video0"], &[]); + let (_track, mut video0) = test.publish("video0"); + let records = segments::MAX_SEGMENTS as u64 + 44; + // The exporter follows from the GOP start; a later join is only restated the tail. + video0.push(&split(0, 0, 40, 0..1)).unwrap(); + test.listed("video0", 0).await; + for sequence in 1..records { + video0 + .push(&split(sequence, sequence * 40, 40, sequence..sequence + 1)) + .unwrap(); + } + let window = test.listed("video0", records - 1).await; + assert_eq!(numbers(&window)[0], 44, "the cap evicted the sync start"); + assert_eq!(advertised(&test.renditions), ["video0"]); + + // A new run starts unadvertised until it lists its own sync start. + test.renditions.fanout().set_generation(Some("a".into())); + test.renditions.fanout().set_generation(Some("b".into())); + video0 + .push(&split(records, records * 40, 40, records..records + 1)) + .unwrap(); + test.listed("video0", records).await; + assert!( + advertised(&test.renditions).is_empty(), + "the latch belongs to the old run" + ); + } + + // Clearing the window, whether a publisher restarted without a new generation (a rewind) or + // records were skipped, drops the advertisement latch with it. + #[tokio::test(start_paused = true)] + async fn a_cleared_window_drops_the_latch() { + let test = Timelines::with(&["video0"], &[]); + let (_track, mut video0) = test.publish("video0"); + video0.push(&split(0, 0, 40, 0..1)).unwrap(); + test.listed("video0", 0).await; + video0.push(&split(1, 40, 40, 1..2)).unwrap(); + test.listed("video0", 1).await; + assert_eq!(advertised(&test.renditions), ["video0"]); + + // The new run opens mid-GOP at an earlier time, under later record numbers. + video0.push(&split(2, 0, 40, 5..6)).unwrap(); + let window = test.listed("video0", 2).await; + assert_eq!(numbers(&window), [2], "the rewind reset the window"); + assert!(advertised(&test.renditions).is_empty(), "a rewind drops the latch"); + + video0.push(&split(3, 40, 40, 0..1)).unwrap(); + test.listed("video0", 3).await; + assert_eq!(advertised(&test.renditions), ["video0"]); + test.renditions.fanout().skip(); + video0.push(&split(4, 80, 40, 1..2)).unwrap(); + let window = test.listed("video0", 4).await; + assert_eq!(numbers(&window), [4], "the skip cleared the window"); + assert!(advertised(&test.renditions).is_empty(), "a skip drops the latch"); + } + + // A span with no media in one rendition keeps its slot as a gap of the same duration there + // alone. The rendition goes on listing after it, at a sync point a player can switch onto, + // and its siblings never notice. + #[tokio::test(start_paused = true)] + async fn a_gap_in_one_rendition_holds_back_nothing() { + let test = Timelines::with(&["video0", "video1"], &["audio0"]); + let (_track0, mut video0) = test.publish("video0"); + let (_track1, mut video1) = test.publish("video1"); + let (_track2, mut audio0) = test.publish("audio0"); + for sequence in 0..8 { + video0.push(&gop(sequence)).unwrap(); + } + // video1 and audio0 have nothing from 4s to 8s (segments 2 and 3). + for (sequence, gop) in [0, 1, 4, 5, 6, 7].into_iter().enumerate() { + let record = span(sequence as u64, gop * 2_000, 2_000, sequence as u64); + video1.push(&record).unwrap(); + audio0.push(&record).unwrap(); + } + + let reference = test.listed("video0", 7).await; + assert!( + reference.segments.iter().all(|s| !s.gap), + "a sibling's gap is not the reference's" + ); + for name in ["video1", "audio0"] { + let snapshot = test.listed(name, 5).await; + let gaps: Vec = snapshot.segments.iter().filter(|s| s.gap).map(|s| s.segment).collect(); + assert_eq!(gaps, [2, 3], "{name}"); + assert!( + snapshot.segments.iter().all(|s| s.duration == Duration::from_secs(2)), + "{name}: a gap keeps its duration" + ); + } + + // Switching to video1 after its gap lands on a segment that starts at a sync point. + let Some(spans::Content::Frames { ranges, .. }) = test.rendition("video1").listed_content(4) else { + panic!("segment 4 resolves on video1"); + }; + assert_eq!( + ranges[0].start, + hang::timeline::Position::group(2), + "its record after the gap" + ); + } } diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index e90623e7db..b95302a125 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -90,6 +90,11 @@ struct Run { init: Option>, /// Bumped by every restart, so an init build that straddles one isn't cached for the new run. epoch: u64, + /// The window [`resets`](segments::Producer::resets) during which a listed segment started + /// where a player can, so the master keeps advertising the rendition even once a GOP past + /// [`MAX_SEGMENTS`](segments::MAX_SEGMENTS) records evicts that segment. Stale once the window + /// is cleared (a new generation, a timeline skip or rewind, a replaced sibling publisher). + startable: Option, } /// The rendition's catalog config, kept whole so a [`Muxer`] can be built per request. @@ -408,7 +413,11 @@ impl Rendition { end: entry.end_time(), discontinuity, }; - self.live.push(row, window); + let starts = self.is(reference) && row.starts_sync(); + let resets = self.live.push(row, window); + if starts { + self.latch(resets); + } self.trim(); } @@ -476,6 +485,7 @@ impl Rendition { generation, init: None, epoch: run.epoch + 1, + startable: None, }; } @@ -635,6 +645,59 @@ impl Rendition { self.live.is_playable() } + /// Whether the multivariant playlist advertises this rendition, i.e. a player can start it. + /// + /// Audio starts on any segment. Video starts only at a group start that is a sync point, + /// so it waits until a listed segment begins at one rather than send a stock player to + /// media it cannot decode. The window keeps its newest such segment, and a GOP too long for + /// that still stays advertised for the rest of the publisher run. + pub(crate) fn is_advertised(&self) -> bool { + self.poll_advertised(&kio::Waiter::noop()).is_ready() + } + + /// Poll until [`is_advertised`](Self::is_advertised), for the master route's long-poll. + pub(crate) fn poll_advertised(&self, waiter: &kio::Waiter) -> Poll<()> { + if self.kind == Kind::Audio { + return self.poll_playable(waiter); + } + self.media.sync(&self.live); + if self.run().startable == Some(self.live.resets()) { + return self.poll_playable(waiter); + } + let startable = self.live.poll_rows(waiter, |rows| { + for row in rows { + let mut content = self.resolve(row); + if content == Content::Pending { + // Wake when the oldest unresolved row resolves, since it may be the sync start. + if self.poll_resolved(waiter, row).is_pending() { + return Poll::Pending; + } + content = self.resolve(row); + } + match content { + // Another rendition's boundary only resolves to frames by snapping to a sync start. + Content::Frames { .. } if !self.is(&row.reference) || row.starts_sync() => return Poll::Ready(()), + Content::Failed => return Poll::Pending, + _ => {} + } + } + Poll::Pending + }); + let Poll::Ready(resets) = startable else { + return Poll::Pending; + }; + // Tagged with the scanned window, so a clear racing this store leaves it stale. + self.latch(resets); + Poll::Ready(()) + } + + /// Record a start listed in window `resets`. Resets only grow, so a delayed latch for an + /// older window never replaces a newer one. + fn latch(&self, resets: u64) { + let mut run = self.run.lock().expect("run lock poisoned"); + run.startable = run.startable.max(Some(resets)); + } + /// This rendition's DASH representation: its master-level metadata plus its track's /// timeline as `(t, d)` pairs in the timeline's own timescale (the record values /// verbatim, so `$Time$` addressing resolves exactly). `None` until the init is known, as @@ -937,6 +1000,12 @@ impl Rendition { self.segment(&tag, segment).await } + /// What the listed segment numbered `segment` holds on this rendition. + pub(crate) fn listed_content(&self, segment: u64) -> Option { + let tag = listed_tag(&self.live.window().segments); + Some(self.resolve(&self.live.row(&tag, segment)?)) + } + /// Fetch the listed segment starting at `time`, under the tag its rows carry. pub(crate) async fn listed_segment_at(&self, time: u64) -> Result> { let tag = listed_tag(&self.live.window().segments); diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index b5253b506a..867f9d383a 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -20,7 +20,7 @@ use moq_mux::timeline::Entry; use super::Upstream; use super::rendition::{Kind, Rendition}; -use super::segments::{Discontinuities, Reference}; +use super::segments::{self, Discontinuities, Reference}; /// The `(kind, name)` identity of a rendition. Video and audio are separate axes, so a video /// and an audio rendition may share a name without colliding. @@ -67,8 +67,9 @@ struct Feed { /// The publisher run every segment URL carries; late-created renditions inherit it. generation: Option>, /// The playlist window duration applied on every push (see - /// [`Config::window`](super::Config::window)), or `None` when the source timeline is - /// authoritative and only its pops trim the playlists. + /// [`Config::window`](super::Config::window)), or `None` in + /// [`Config::history`](super::Config::history) mode, where only the source timeline's pops + /// trim the playlists. window: Option, } @@ -218,13 +219,19 @@ impl Fanout { feed.history .push_back((index, entry.clone(), reference.clone(), discontinuity)); let window = feed.window; - while let Some(window) = window - && feed.history.len() >= 2 - { - let newest = &feed.history.back().unwrap().1; - let span = - (Duration::from(newest.pts) + newest.duration).saturating_sub(Duration::from(feed.history[1].1.pts)); - if span < window { + while let Some(window) = window { + let rest = match feed.history.get(1) { + Some((_, second, ..)) => { + let newest = &feed.history.back().unwrap().1; + newest.end_time().saturating_sub(Duration::from(second.pts)) + } + None => Duration::ZERO, + }; + let starts = |(_, entry, reference, _): &(u64, Entry, Reference, u64)| { + segments::starts(reference, entry.keyframe, &entry.start) + }; + let keeps_start = !starts(&feed.history[0]) || feed.history.iter().skip(1).any(starts); + if !segments::evicts(window, feed.history.len(), rest, keeps_start) { break; } feed.history.pop_front(); @@ -343,6 +350,25 @@ impl Producer { .await; } + /// Resolve once at least one rendition is [advertised](Rendition::is_advertised), so a master + /// playlist can list it. + #[cfg_attr(not(feature = "server"), allow(dead_code))] + pub(crate) async fn advertised(&self) { + let _ = kio::wait(|waiter| { + self.state.poll_ref(waiter, |current| { + if current + .values() + .any(|rendition| rendition.poll_advertised(waiter).is_ready()) + { + Poll::Ready(()) + } else { + Poll::Pending + } + }) + }) + .await; + } + /// Enroll a freshly-created rendition in the timeline feed, replaying the recent history /// (and the ended/closed markers) so its window matches its siblings', and start following its /// own timeline. diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index f18aba26c6..934fc785df 100644 --- a/rs/moq-hls/src/export/segments.rs +++ b/rs/moq-hls/src/export/segments.rs @@ -45,6 +45,8 @@ struct State { sequence: u64, /// The timeline track ended: the broadcast is over (`EXT-X-ENDLIST`). ended: bool, + /// Bumped whenever the window is cleared, so a latch on what it listed can tell it is stale. + resets: u64, } /// Numbers the broadcast's content timeline: the sequence bumps wherever the timeline breaks. @@ -82,6 +84,28 @@ impl Discontinuities { } } +/// The most segments a playlist window lists, however short they are. +/// +/// A fresh subscriber to a `moq-mux` timeline is only promised this many recent records (the +/// checkpoint each timeline group restates), so an edge that joined long ago must not list more +/// than one joining now, or the two would disagree on `EXT-X-MEDIA-SEQUENCE`. The dense-timeline +/// test fails if the two drift apart. +pub(crate) const MAX_SEGMENTS: usize = 256; + +/// Whether a window of `len` rows, whose rows after the oldest span `rest`, should drop its +/// oldest: it lists more than [`MAX_SEGMENTS`], or the rest still covers `window` and still +/// holds a start (`keeps_start`, see [`starts`]). A GOP longer than the window keeps its sync +/// row, so the window stretches to that one GOP, and stays startable. +pub(crate) fn evicts(window: Duration, len: usize, rest: Duration, keeps_start: bool) -> bool { + len > MAX_SEGMENTS || (len >= 2 && rest >= window && keeps_start) +} + +/// Whether a player can start at a `reference` record: audio starts anywhere, video only at a +/// group start that is a sync point. +pub(crate) fn starts(reference: &(Kind, String), keyframe: bool, start: &hang::timeline::Position) -> bool { + reference.0 == Kind::Audio || (keyframe && start.frame == 0) +} + /// The rendition a broadcast's segment boundaries come from. pub(crate) type Reference = Arc<(Kind, String)>; @@ -104,7 +128,8 @@ pub(crate) struct Row { pub reference: Reference, /// The reference record's frames, which the reference rendition serves as is. pub frames: std::ops::Range, - /// Whether the reference record starts on a keyframe. + /// Whether the reference record's first frame is a sync point, where a player can start + /// decoding (the record's `keyframe` flag). pub keyframe: bool, /// Presentation duration. pub duration: Duration, @@ -118,6 +143,19 @@ pub(crate) struct Row { pub discontinuity: u64, } +impl Row { + /// Whether the segment starts at a group start that is a sync point, so a player can begin + /// decoding with it. + pub fn starts_sync(&self) -> bool { + self.keyframe && self.frames.start.frame == 0 + } + + /// Whether a player can start at this segment (see [`starts`]). + fn starts(&self) -> bool { + starts(&self.reference, self.keyframe, &self.frames.start) + } +} + /// A consistent read of the window, for rendering one playlist (the serve path only). #[cfg_attr(not(feature = "server"), allow(dead_code))] pub(crate) struct Window { @@ -178,15 +216,17 @@ impl Producer { rows: VecDeque::new(), sequence: 0, ended: false, + resets: 0, }), } } - /// Append a row, evicting the front of the window past `window`. With no `window`, only - /// source timeline pops remove rows. - pub fn push(&self, row: Row, window: Option) { + /// Append a row, evicting the front of the window past `window` (see [`evicts`]). With no + /// `window`, only source timeline pops remove rows. Returns the window's + /// [`resets`](Self::resets) after the push. + pub fn push(&self, row: Row, window: Option) -> u64 { let Ok(mut state) = self.state.write() else { - return; + return u64::MAX; }; if let Some(back) = state.rows.back() { @@ -195,6 +235,7 @@ impl Producer { if Duration::from(row.pts) < Duration::from(back.pts) || row.segment <= back.segment { tracing::warn!("timeline jumped backwards; resetting the playlist window"); state.rows.clear(); + state.resets += 1; } } if state.rows.is_empty() { @@ -203,17 +244,19 @@ impl Producer { state.rows.push_back(row); - // Evict from the front while the remaining rows still cover the window. - while let Some(window) = window - && state.rows.len() >= 2 - { - let span = state.rows.back().unwrap().end.saturating_sub(state.rows[1].pts.into()); - if span < window { + while let Some(window) = window { + let rest = match state.rows.get(1) { + Some(second) => state.rows.back().unwrap().end.saturating_sub(second.pts.into()), + None => Duration::ZERO, + }; + let keeps_start = !state.rows[0].starts() || state.rows.iter().skip(1).any(Row::starts); + if !evicts(window, state.rows.len(), rest, keeps_start) { break; } state.rows.pop_front(); } state.sequence = state.rows.front().unwrap().segment; + state.resets } /// Remove rows whose source timeline indices fall within `range`. @@ -239,9 +282,16 @@ impl Producer { pub fn clear(&self) { if let Ok(mut state) = self.state.write() { state.rows.clear(); + state.resets += 1; } } + /// How many times the window was cleared: by [`clear`](Self::clear), or a timeline that + /// jumped backwards. + pub fn resets(&self) -> u64 { + self.state.read().resets + } + /// Mark the timeline ended (it finished, or failed): the playlist gets /// `EXT-X-ENDLIST` and cursors end once drained. pub fn end(&self) { @@ -302,7 +352,7 @@ impl Producer { .iter() .rev() .filter(|row| *row.reference == *reference) - .find(|row| row.keyframe && row.frames.start.frame == 0) + .find(|row| row.starts_sync()) .map(|row| row.frames.start.group) } @@ -331,6 +381,19 @@ impl Producer { } } + /// Poll `f` over the listed rows, waking on the next window change while it is pending, and + /// return the [`resets`](Self::resets) those rows belong to. A closed window stays pending: no + /// new row can make `f` ready. + pub fn poll_rows(&self, waiter: &kio::Waiter, mut f: impl FnMut(&VecDeque) -> Poll<()>) -> Poll { + match self + .state + .poll_ref(waiter, |state| f(&state.rows).map(|()| state.resets)) + { + Poll::Ready(Ok(resets)) => Poll::Ready(resets), + _ => Poll::Pending, + } + } + /// A cursor over segments, starting from the oldest still in the window. pub fn subscribe(&self, rendition: Arc) -> Consumer { Consumer { @@ -589,16 +652,16 @@ mod tests { fn backwards_jump_resets_the_window() { let live = Producer::new(); let window = Some(Duration::from_secs(30)); - live.push(row(0, 0, 10_000, 2_000), window); - live.push(row(1, 1, 12_000, 2_000), window); - live.push(row(2, 2, 1_000, 2_000), window); // restart: pts rewound + assert_eq!(live.push(row(0, 0, 10_000, 2_000), window), 0); + assert_eq!(live.push(row(1, 1, 12_000, 2_000), window), 0); + assert_eq!(live.push(row(2, 2, 1_000, 2_000), window), 1); // restart: pts rewound let snapshot = live.window(); assert_eq!(snapshot.segments.len(), 1, "the window restarted at the new row"); assert_eq!(snapshot.segments[0].segment, 2); // A segment number that rewinds (a restarted publisher) resets the same way. - live.push(row(0, 0, 2_000, 2_000), window); + assert_eq!(live.push(row(0, 0, 2_000, 2_000), window), 2); assert_eq!(live.window().segments.len(), 1); } diff --git a/rs/moq-hls/src/export/spans.rs b/rs/moq-hls/src/export/spans.rs index fcc7d17973..8fd1a2fb23 100644 --- a/rs/moq-hls/src/export/spans.rs +++ b/rs/moq-hls/src/export/spans.rs @@ -167,11 +167,17 @@ impl Spans { .records .iter() .rev() - .find(|record| record.keyframe && record.start.frame == 0) + .find(|record| starts_sync(record)) .map(|record| record.start.group) } } +/// Whether `record` starts at a group start that is a sync point, so a player can begin decoding +/// there. +fn starts_sync(record: &Entry) -> bool { + record.keyframe && record.start.frame == 0 +} + /// Snap `boundary` to the nearest keyframe record start within [`TOLERANCE`], preferring the /// earlier one on a tie. fn snap<'a>(candidates: &[&'a Entry], boundary: Duration) -> Option<&'a Entry> { @@ -206,11 +212,7 @@ fn video(records: Vec<&Entry>, ended: bool, span: Range) -> Content { return Content::Pending; } - let candidates: Vec<&Entry> = records - .iter() - .copied() - .filter(|record| record.keyframe && record.start.frame == 0) - .collect(); + let candidates: Vec<&Entry> = records.iter().copied().filter(|record| starts_sync(record)).collect(); let Some(first) = snap(&candidates, span.start) else { return Content::Gap; }; diff --git a/rs/moq-hls/src/server/routes.rs b/rs/moq-hls/src/server/routes.rs index f2dd64ce2b..024f13e86c 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -234,11 +234,14 @@ async fn master(server: &Server, broadcast: &str, query: Option<&str>) -> Respon let Some(broadcaster) = server.broadcaster(broadcast).await else { return not_found(); }; - let _ = tokio::time::timeout(READY_TIMEOUT, broadcaster.ready()).await; - if broadcaster.is_empty() { + // A master listing no variant is unplayable: wait for one that can start, else answer + // unavailable. + let _ = tokio::time::timeout(READY_TIMEOUT, broadcaster.advertised()).await; + let (video, audio) = broadcaster.variants(query); + if video.is_empty() && audio.is_empty() { return not_found(); } - m3u8(broadcaster.master_playlist(query)) + m3u8(crate::export::master::render(&video, &audio)) } async fn manifest(server: &Server, broadcast: &str, query: Option<&str>) -> Response { @@ -655,6 +658,22 @@ mod tests { oneshot(app.clone(), uri).await.status() } + /// A master listing no variant would be unplayable, so it answers 404 until a rendition can + /// start. + #[tokio::test(start_paused = true)] + async fn master_answers_404_until_a_rendition_can_start() { + let origin = moq_tokio::origin::spawn(); + let mut broadcast = origin.create_broadcast("live").expect("publish"); + broadcast.announce(Default::default()).expect("announce"); + let (_catalog, _registration, _track, mut media) = publish_video(&mut broadcast, video_config(), None); + + let app = Server::new(origin.consume(), crate::export::Config::default()).router(); + assert_eq!(status(&app, "/live/master.m3u8").await, StatusCode::NOT_FOUND); + + write_three_gops(&mut media); + wait_listed(&app, "/live/master.m3u8", "video/video0/media.m3u8").await; + } + /// Only the URLs the renderers emit are served: the init under its hash, segments under the /// current generation. #[tokio::test]