From ebdea46872d554c8f50ffcbb7caa61e9d5effbcb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:07:00 -0700 Subject: [PATCH 1/9] quest: claim quest/m1/hls-bounded From dccb6d750c46be79585eee9075526151880f101f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:26:52 -0700 Subject: [PATCH 2/9] feat(hls)!: bounded playlists: capped window, sync-point gating, explicit replay mode A durable timeline now lists the same capped window as a live one; listing past it is the explicit export::Config::replay mode (reverses #4155). The window lists at most 256 segments, so an edge that followed a dense timeline agrees with one joining now. The master advertises a video rendition only once a listed segment starts at a group start that is a sync point, and answers 404 while no rendition can start. Co-Authored-By: Claude Opus 5.5 --- doc/bin/hls.md | 25 +- quest/m1/README.md | 1 - quest/m1/archive/README.md | 10 +- quest/m1/archive/dvr.md | 4 +- quest/m1/archive/replay-catalog.md | 12 +- quest/m1/hls-bounded.md | 87 ------- rs/moq-cli/src/hls.rs | 6 +- rs/moq-hls/src/export/archive_tests.rs | 42 +++- rs/moq-hls/src/export/mod.rs | 324 ++++++++++++++++++++++--- rs/moq-hls/src/export/rendition.rs | 26 ++ rs/moq-hls/src/export/renditions.rs | 23 +- rs/moq-hls/src/export/segments.rs | 43 +++- rs/moq-hls/src/export/spans.rs | 14 +- rs/moq-hls/src/server/routes.rs | 22 +- 14 files changed, 464 insertions(+), 175 deletions(-) delete mode 100644 quest/m1/hls-bounded.md diff --git a/doc/bin/hls.md b/doc/bin/hls.md index c1b86864fa..e7d23b46cf 100644 --- a/doc/bin/hls.md +++ b/doc/bin/hls.md @@ -59,20 +59,33 @@ 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::replay` 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`. 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. + 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 21d2307fea..ec90833f28 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -88,7 +88,6 @@ blocks. The quests that gated m0 lines moved under them. - [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 - [Request update routing](/quest/m1/ietf-subscribe-update-route.md) - drafts 14 and 15 SUBSCRIBE_UPDATE and draft 16 REQUEST_UPDATE reach the request named by their second field - [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 diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index feb8ad71e6..b05026c418 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::replay` 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..413ffb5305 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::replay` 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..70a6779f6a 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 replay 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::replay` mode, which `moq +export hls` does not expose yet. Give it a flag here. Replay 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..06b610a651 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}"); +} + +/// Replay 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 replay_mode_lists_a_durable_timeline_past_the_window() { + let recording = segments(12).await; + let config = Config { + replay: 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..c5b7b62fc2 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -70,19 +70,32 @@ 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, and past 256 + /// segments 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 [`replay`](Self::replay) 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. + pub replay: bool, } impl Default for Config { fn default() -> Self { Self { window: Duration::from_secs(16), + replay: false, } } } @@ -123,7 +136,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.replay)); *broadcaster.watcher.lock().unwrap() = Some(watcher); Ok(broadcaster) } @@ -188,11 +201,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 +234,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 +348,7 @@ async fn watch_catalog( upstream: Upstream, renditions: renditions::Producer, timeline_watcher: Arc>>>, + replay: 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 +379,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 replay && !unbound && catalog.archive.as_ref().is_some_and(durable) { unbound = true; renditions.fanout().unbound(); } @@ -417,8 +441,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::replay`] 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 +2876,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 +2955,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 +2996,7 @@ mod tests { } } - impl Drop for Failing { + impl Drop for Timelines { fn drop(&mut self) { self.watcher.abort(); } @@ -2958,7 +3016,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 +3046,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 +3074,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 +3097,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 +3142,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 +3157,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 +3167,202 @@ 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!(read <= bound, "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"]); + + for sequence in 4..7 { + video0.push(&gop(sequence)).unwrap(); + video1.push(&gop(sequence)).unwrap(); + audio0.push(&gop(sequence)).unwrap(); + } + test.listed("video1", 4).await; + assert_eq!(advertised(&test.renditions), ["video0", "video1", "audio0"]); + } + + // 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..c97bf9717c 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -635,6 +635,26 @@ 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. + pub(crate) fn is_advertised(&self) -> bool { + if !self.is_playable() { + return false; + } + if self.kind == Kind::Audio { + return true; + } + let window = self.live.window(); + self.resolved(&window.segments).any(|(row, content)| match content { + // Another rendition's boundary only resolves to frames by snapping to a sync start. + Content::Frames { .. } => !self.is(&row.reference) || row.starts_sync(), + _ => false, + }) + } + /// 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 +957,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..83b49289fd 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::replay`](super::Config::replay) mode, where only the source timeline's pops + /// trim the playlists. window: Option, } @@ -218,13 +219,15 @@ 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, + }; + if !segments::evicts(window, feed.history.len(), rest) { break; } feed.history.pop_front(); diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index f18aba26c6..bec36e59f3 100644 --- a/rs/moq-hls/src/export/segments.rs +++ b/rs/moq-hls/src/export/segments.rs @@ -82,6 +82,20 @@ 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`. +pub(crate) fn evicts(window: Duration, len: usize, rest: Duration) -> bool { + len > MAX_SEGMENTS || (len >= 2 && rest >= window) +} + /// The rendition a broadcast's segment boundaries come from. pub(crate) type Reference = Arc<(Kind, String)>; @@ -104,7 +118,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 +133,14 @@ 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 + } +} + /// 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 { @@ -182,8 +205,8 @@ impl Producer { } } - /// Append a row, evicting the front of the window past `window`. With no `window`, only - /// source timeline pops remove rows. + /// Append a row, evicting the front of the window past `window` (see [`evicts`]). With no + /// `window`, only source timeline pops remove rows. pub fn push(&self, row: Row, window: Option) { let Ok(mut state) = self.state.write() else { return; @@ -203,12 +226,12 @@ 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, + }; + if !evicts(window, state.rows.len(), rest) { break; } state.rows.pop_front(); @@ -302,7 +325,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) } 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..6eafb5afb0 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -235,10 +235,12 @@ async fn master(server: &Server, broadcast: &str, query: Option<&str>) -> Respon return not_found(); }; let _ = tokio::time::timeout(READY_TIMEOUT, broadcaster.ready()).await; - if broadcaster.is_empty() { + // A master listing no variant is unplayable: answer unavailable until one can start. + 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 +657,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] From 5380383cba5851256e785db61b3c3cf8f3f6503a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 12:25:53 -0700 Subject: [PATCH 3/9] refactor(hls): rename Config::replay to Config::history Avoids clashing with the catalog's archive.replay path. Co-Authored-By: Claude Opus 5.5 --- doc/bin/hls.md | 2 +- quest/m1/archive/README.md | 2 +- quest/m1/archive/dvr.md | 2 +- quest/m1/archive/replay-catalog.md | 6 +++--- rs/moq-hls/src/export/archive_tests.rs | 6 +++--- rs/moq-hls/src/export/mod.rs | 14 +++++++------- rs/moq-hls/src/export/renditions.rs | 2 +- 7 files changed, 17 insertions(+), 17 deletions(-) diff --git a/doc/bin/hls.md b/doc/bin/hls.md index e7d23b46cf..cbf90966bd 100644 --- a/doc/bin/hls.md +++ b/doc/bin/hls.md @@ -61,7 +61,7 @@ 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. A recording's playlists list the same capped window as a live broadcast. A library embedder can set -`export::Config::replay` to list past it instead: when the catalog's `archive` +`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 diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index b05026c418..d5aad63e07 100644 --- a/quest/m1/archive/README.md +++ b/quest/m1/archive/README.md @@ -50,7 +50,7 @@ Every track has its own timeline (`rs/moq-mux/src/timeline.rs`): objects. The caller supplies the catalog. - A catalog `archive` entry with a `store` and no `replay` path declares its spans durable on that broadcast. The exporter still lists a capped window - for it; only the explicit `export::Config::replay` mode lists the whole + 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 diff --git a/quest/m1/archive/dvr.md b/quest/m1/archive/dvr.md index 413ffb5305..367c01531b 100644 --- a/quest/m1/archive/dvr.md +++ b/quest/m1/archive/dvr.md @@ -44,7 +44,7 @@ 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. 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::replay` listing. `moq-hls` gains +`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 70a6779f6a..5e1e62e139 100644 --- a/quest/m1/archive/replay-catalog.md +++ b/quest/m1/archive/replay-catalog.md @@ -6,7 +6,7 @@ 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 an exporter in replay mode treats its timeline as durable and lists past +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 @@ -37,8 +37,8 @@ this by hand-building a catalog. 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::replay` mode, which `moq -export hls` does not expose yet. Give it a flag here. Replay mode lists from +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 diff --git a/rs/moq-hls/src/export/archive_tests.rs b/rs/moq-hls/src/export/archive_tests.rs index 06b610a651..6a0db647df 100644 --- a/rs/moq-hls/src/export/archive_tests.rs +++ b/rs/moq-hls/src/export/archive_tests.rs @@ -584,14 +584,14 @@ async fn a_durable_timeline_lists_the_window() { assert!(!playlist.contains("seg/29b1cd67.3.m4s\n"), "{playlist}"); } -/// Replay mode lists a durable timeline's whole recording, and DASH offers the whole listed +/// 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 replay_mode_lists_a_durable_timeline_past_the_window() { +async fn history_mode_lists_a_durable_timeline_past_the_window() { let recording = segments(12).await; let config = Config { - replay: true, + history: true, ..Config::default() }; let replay = Replay::open_with(&recording, 64 * 1024 * 1024, durable(), config.clone()).await; diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index c5b7b62fc2..ee365d4a47 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -75,7 +75,7 @@ pub struct Config { /// segments are fetched from there on request. It also caps segment `Cache-Control: max-age`. /// /// A durable timeline gets the same window: a live playlist stays bounded however much its - /// store retains. See [`replay`](Self::replay) to list all of it. + /// 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 @@ -88,14 +88,14 @@ pub struct Config { /// 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. - pub replay: bool, + pub history: bool, } impl Default for Config { fn default() -> Self { Self { window: Duration::from_secs(16), - replay: false, + history: false, } } } @@ -136,7 +136,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, config.replay)); + let watcher = tokio::spawn(watch_catalog(upstream, renditions, timeline_watcher, config.history)); *broadcaster.watcher.lock().unwrap() = Some(watcher); Ok(broadcaster) } @@ -348,7 +348,7 @@ async fn watch_catalog( upstream: Upstream, renditions: renditions::Producer, timeline_watcher: Arc>>>, - replay: bool, + 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 @@ -379,7 +379,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 replay && !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(); } @@ -441,7 +441,7 @@ 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 [`Config::replay`] can follow the +/// 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() diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index 83b49289fd..34a831bf8c 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -68,7 +68,7 @@ struct Feed { generation: Option>, /// The playlist window duration applied on every push (see /// [`Config::window`](super::Config::window)), or `None` in - /// [`Config::replay`](super::Config::replay) mode, where only the source timeline's pops + /// [`Config::history`](super::Config::history) mode, where only the source timeline's pops /// trim the playlists. window: Option, } From 47b97f0580edd7a935460e6ab129e6449978d078 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 16:22:22 -0700 Subject: [PATCH 4/9] fix(hls): master long-polls until a rendition can start; pin the fresh-join lower bound The master route waited on any playable rendition, then applied the stricter sync gate, so a window starting mid-GOP answered 404 at once instead of waiting. It now waits on the gate itself. Also document that history-mode edges can disagree on EXT-X-MEDIA-SEQUENCE, and make the fresh-join test fail if it reads nothing. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 32 +++++++++++++++++++++++++-- rs/moq-hls/src/export/rendition.rs | 34 +++++++++++++++++++++-------- rs/moq-hls/src/export/renditions.rs | 19 ++++++++++++++++ rs/moq-hls/src/export/segments.rs | 9 ++++++++ rs/moq-hls/src/server/routes.rs | 5 +++-- 5 files changed, 86 insertions(+), 13 deletions(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index ee365d4a47..91c782b3de 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -87,7 +87,9 @@ pub struct Config { /// /// 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. + /// 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, } @@ -170,6 +172,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). /// @@ -3222,7 +3230,10 @@ mod tests { .unwrap(); } let read = fresh_join_reads(&early, "video0").await; - assert!(read <= bound, "day {day}: a fresh join read {read} records"); + assert!( + (segments::MAX_SEGMENTS..=bound).contains(&read), + "day {day}: a fresh join read {read} records" + ); } let last = 3 * DAY - 1; @@ -3312,11 +3323,28 @@ mod tests { 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"]); } diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index c97bf9717c..b51546a660 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -641,17 +641,33 @@ impl Rendition { /// so it waits until a listed segment begins at one rather than send a stock player to /// media it cannot decode. pub(crate) fn is_advertised(&self) -> bool { - if !self.is_playable() { - return false; - } + 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 true; + return self.poll_playable(waiter); } - let window = self.live.window(); - self.resolved(&window.segments).any(|(row, content)| match content { - // Another rendition's boundary only resolves to frames by snapping to a sync start. - Content::Frames { .. } => !self.is(&row.reference) || row.starts_sync(), - _ => false, + self.media.sync(&self.live); + 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 }) } diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index 34a831bf8c..eab5d02ea2 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -346,6 +346,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 bec36e59f3..6dc691b33c 100644 --- a/rs/moq-hls/src/export/segments.rs +++ b/rs/moq-hls/src/export/segments.rs @@ -354,6 +354,15 @@ impl Producer { } } + /// Poll `f` over the listed rows, waking on the next window change while it is pending. 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)) { + Poll::Ready(Ok(())) => Poll::Ready(()), + _ => Poll::Pending, + } + } + /// A cursor over segments, starting from the oldest still in the window. pub fn subscribe(&self, rendition: Arc) -> Consumer { Consumer { diff --git a/rs/moq-hls/src/server/routes.rs b/rs/moq-hls/src/server/routes.rs index 6eafb5afb0..024f13e86c 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -234,8 +234,9 @@ 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; - // A master listing no variant is unplayable: answer unavailable until one can start. + // 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(); From ce044d590ed7f36c08a76b851de3188e50fb840e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 16:51:36 -0700 Subject: [PATCH 5/9] fix(hls): keep a long GOP's sync start in the window; latch advertisement per run The window never evicts its newest segment a video player can start at, so a GOP longer than the window stretches it instead of leaving nothing startable. A GOP past the 256-record cap still loses that segment, so a rendition stays advertised for the rest of its publisher run once one was listed. Co-Authored-By: Claude Opus 5.5 --- doc/bin/hls.md | 9 ++++- rs/moq-hls/src/export/mod.rs | 63 ++++++++++++++++++++++++++++- rs/moq-hls/src/export/rendition.rs | 21 ++++++++-- rs/moq-hls/src/export/renditions.rs | 6 ++- rs/moq-hls/src/export/segments.rs | 22 ++++++++-- 5 files changed, 108 insertions(+), 13 deletions(-) diff --git a/doc/bin/hls.md b/doc/bin/hls.md index cbf90966bd..94b21460da 100644 --- a/doc/bin/hls.md +++ b/doc/bin/hls.md @@ -73,11 +73,16 @@ 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`. A rendition with no media for a span lists that span as +`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 one GOP plus `--window` 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. +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 diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 91c782b3de..642bcf2115 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -70,8 +70,9 @@ 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, and past 256 - /// segments however short they are. Keep it within the relay's group-cache retention, since + /// 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 gets the same window: a live playlist stays bounded however much its @@ -3349,6 +3350,64 @@ mod tests { 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"); + 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 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. diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index b51546a660..703959eaf9 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -90,6 +90,10 @@ struct Run { init: Option>, /// Bumped by every restart, so an init build that straddles one isn't cached for the new run. epoch: u64, + /// A listed segment of this run 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. + startable: bool, } /// The rendition's catalog config, kept whole so a [`Muxer`] can be built per request. @@ -408,6 +412,9 @@ impl Rendition { end: entry.end_time(), discontinuity, }; + if self.is(reference) && row.starts_sync() { + self.run.lock().expect("run lock poisoned").startable = true; + } self.live.push(row, window); self.trim(); } @@ -476,6 +483,7 @@ impl Rendition { generation, init: None, epoch: run.epoch + 1, + startable: false, }; } @@ -639,18 +647,19 @@ impl Rendition { /// /// 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. + /// 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 { + if self.kind == Kind::Audio || self.run().startable { return self.poll_playable(waiter); } self.media.sync(&self.live); - self.live.poll_rows(waiter, |rows| { + let startable = self.live.poll_rows(waiter, |rows| { for row in rows { let mut content = self.resolve(row); if content == Content::Pending { @@ -668,7 +677,11 @@ impl Rendition { } } Poll::Pending - }) + }); + if startable.is_ready() { + self.run.lock().expect("run lock poisoned").startable = true; + } + startable } /// This rendition's DASH representation: its master-level metadata plus its track's diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index eab5d02ea2..867f9d383a 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -227,7 +227,11 @@ impl Fanout { } None => Duration::ZERO, }; - if !segments::evicts(window, feed.history.len(), rest) { + 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(); diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index 6dc691b33c..29ef21634d 100644 --- a/rs/moq-hls/src/export/segments.rs +++ b/rs/moq-hls/src/export/segments.rs @@ -91,9 +91,17 @@ impl Discontinuities { 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`. -pub(crate) fn evicts(window: Duration, len: usize, rest: Duration) -> bool { - len > MAX_SEGMENTS || (len >= 2 && rest >= window) +/// 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 can grow to one GOP plus `window`, 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. @@ -139,6 +147,11 @@ impl Row { 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). @@ -231,7 +244,8 @@ impl Producer { Some(second) => state.rows.back().unwrap().end.saturating_sub(second.pts.into()), None => Duration::ZERO, }; - if !evicts(window, state.rows.len(), rest) { + 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(); From d791dceec18d96b80d4b07ee7918c36f743f1579 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 17:56:31 -0700 Subject: [PATCH 6/9] fix(hls): a backward timeline jump resets the advertisement latch; tighten the long-GOP bound Co-Authored-By: Claude Opus 5.5 --- doc/bin/hls.md | 4 ++-- rs/moq-hls/src/export/mod.rs | 14 ++++++++++++++ rs/moq-hls/src/export/rendition.rs | 9 +++++---- rs/moq-hls/src/export/segments.rs | 12 ++++++++---- 4 files changed, 29 insertions(+), 10 deletions(-) diff --git a/doc/bin/hls.md b/doc/bin/hls.md index 94b21460da..85b0688e1d 100644 --- a/doc/bin/hls.md +++ b/doc/bin/hls.md @@ -75,8 +75,8 @@ 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 one GOP plus `--window` rather than leave nothing a -player can start at. A rendition with no media for a span lists that span as +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 diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 642bcf2115..27ff37a392 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -3379,6 +3379,8 @@ mod tests { } 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(); @@ -3406,6 +3408,18 @@ mod tests { 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" + ); } // A span with no media in one rendition keeps its slot as a gap of the same duration there diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 703959eaf9..6ace622e2a 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -92,7 +92,7 @@ struct Run { epoch: u64, /// A listed segment of this run 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. + /// that segment. A new generation or a timeline that jumps backwards starts a new run. startable: bool, } @@ -412,10 +412,11 @@ impl Rendition { end: entry.end_time(), discontinuity, }; - if self.is(reference) && row.starts_sync() { - self.run.lock().expect("run lock poisoned").startable = true; + let starts = self.is(reference) && row.starts_sync(); + let restarted = self.live.push(row, window); + if starts || restarted { + self.run.lock().expect("run lock poisoned").startable = starts; } - self.live.push(row, window); self.trim(); } diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index 29ef21634d..56e354395c 100644 --- a/rs/moq-hls/src/export/segments.rs +++ b/rs/moq-hls/src/export/segments.rs @@ -93,7 +93,7 @@ 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 can grow to one GOP plus `window`, and stays startable. +/// 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) } @@ -219,11 +219,13 @@ impl Producer { } /// Append a row, evicting the front of the window past `window` (see [`evicts`]). With no - /// `window`, only source timeline pops remove rows. - pub fn push(&self, row: Row, window: Option) { + /// `window`, only source timeline pops remove rows. Returns whether the timeline jumped + /// backwards, which starts a new publisher run. + pub fn push(&self, row: Row, window: Option) -> bool { let Ok(mut state) = self.state.write() else { - return; + return false; }; + let mut restarted = false; if let Some(back) = state.rows.back() { // A backward jump in pts or segment number means the publisher restarted its timeline; @@ -231,6 +233,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(); + restarted = true; } } if state.rows.is_empty() { @@ -251,6 +254,7 @@ impl Producer { state.rows.pop_front(); } state.sequence = state.rows.front().unwrap().segment; + restarted } /// Remove rows whose source timeline indices fall within `range`. From edf64664fcc3f323345e0e56611f5294c547ce44 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 18:18:25 -0700 Subject: [PATCH 7/9] fix(hls): serialize the advertisement latch with window resets; pin the rewind reset poll_advertised scanned the window, released it, then latched, so a reset in between could carry the old run's latch into the new one. The run lock now spans the scan and the push. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 22 ++++++++++++++++++++++ rs/moq-hls/src/export/rendition.rs | 15 ++++++++++----- rs/moq-hls/src/export/segments.rs | 10 +++++----- 3 files changed, 37 insertions(+), 10 deletions(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 27ff37a392..ccf2be5ec9 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -3422,6 +3422,28 @@ mod tests { ); } + // A publisher restarted without a new generation rewinds the timeline, which resets the + // advertisement latch with the window. + #[tokio::test(start_paused = true)] + async fn a_rewound_timeline_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(), + "the latch belongs to the old run" + ); + } + // 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. diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 6ace622e2a..456493ac82 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -92,7 +92,8 @@ struct Run { epoch: u64, /// A listed segment of this run 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. A new generation or a timeline that jumps backwards starts a new run. + /// that segment. A new generation, or a timeline that jumps backwards, resets it with the + /// window. startable: bool, } @@ -413,10 +414,12 @@ impl Rendition { discontinuity, }; let starts = self.is(reference) && row.starts_sync(); - let restarted = self.live.push(row, window); - if starts || restarted { - self.run.lock().expect("run lock poisoned").startable = starts; + // Held across the push, so a concurrent `poll_advertised` can't latch a cleared window. + let mut run = self.run.lock().expect("run lock poisoned"); + if self.live.push(row, window) || starts { + run.startable = starts; } + drop(run); self.trim(); } @@ -660,6 +663,8 @@ impl Rendition { return self.poll_playable(waiter); } self.media.sync(&self.live); + // Held across the scan, so a reset (`push`, `restart`) can't land between it and the latch. + let mut run = self.run.lock().expect("run lock poisoned"); let startable = self.live.poll_rows(waiter, |rows| { for row in rows { let mut content = self.resolve(row); @@ -680,7 +685,7 @@ impl Rendition { Poll::Pending }); if startable.is_ready() { - self.run.lock().expect("run lock poisoned").startable = true; + run.startable = true; } startable } diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index 56e354395c..6fbf437a77 100644 --- a/rs/moq-hls/src/export/segments.rs +++ b/rs/moq-hls/src/export/segments.rs @@ -220,7 +220,7 @@ impl Producer { /// Append a row, evicting the front of the window past `window` (see [`evicts`]). With no /// `window`, only source timeline pops remove rows. Returns whether the timeline jumped - /// backwards, which starts a new publisher run. + /// backwards, which resets the window. pub fn push(&self, row: Row, window: Option) -> bool { let Ok(mut state) = self.state.write() else { return false; @@ -639,16 +639,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!(!live.push(row(0, 0, 10_000, 2_000), window)); + assert!(!live.push(row(1, 1, 12_000, 2_000), window)); + assert!(live.push(row(2, 2, 1_000, 2_000), window)); // 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!(live.push(row(0, 0, 2_000, 2_000), window)); assert_eq!(live.window().segments.len(), 1); } From 368fdd20f5417ceabc42a0262ec2b06146898158 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 18:24:37 -0700 Subject: [PATCH 8/9] fix(hls): any window clear drops the advertisement latch The latch now records which window it saw a start in, so every clear (a new generation, a skip, a rewind, a replaced sibling) leaves it stale, and a clear racing the latch can't carry it over without cross-lock coordination. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/mod.rs | 20 ++++++++----- rs/moq-hls/src/export/rendition.rs | 37 ++++++++++++------------ rs/moq-hls/src/export/segments.rs | 45 +++++++++++++++++++----------- 3 files changed, 61 insertions(+), 41 deletions(-) diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index ccf2be5ec9..c0a07b11d0 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -3422,10 +3422,10 @@ mod tests { ); } - // A publisher restarted without a new generation rewinds the timeline, which resets the - // advertisement latch with the window. + // 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_rewound_timeline_drops_the_latch() { + 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(); @@ -3438,10 +3438,16 @@ mod tests { 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(), - "the latch belongs to the old run" - ); + 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 diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 456493ac82..26927da1b8 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -90,11 +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, - /// A listed segment of this run 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. A new generation, or a timeline that jumps backwards, resets it with the - /// window. - startable: bool, + /// 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. @@ -414,12 +414,10 @@ impl Rendition { discontinuity, }; let starts = self.is(reference) && row.starts_sync(); - // Held across the push, so a concurrent `poll_advertised` can't latch a cleared window. - let mut run = self.run.lock().expect("run lock poisoned"); - if self.live.push(row, window) || starts { - run.startable = starts; + let resets = self.live.push(row, window); + if starts { + self.run.lock().expect("run lock poisoned").startable = Some(resets); } - drop(run); self.trim(); } @@ -487,7 +485,7 @@ impl Rendition { generation, init: None, epoch: run.epoch + 1, - startable: false, + startable: None, }; } @@ -659,12 +657,13 @@ impl Rendition { /// 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 || self.run().startable { + if self.kind == Kind::Audio { return self.poll_playable(waiter); } self.media.sync(&self.live); - // Held across the scan, so a reset (`push`, `restart`) can't land between it and the latch. - let mut run = self.run.lock().expect("run lock poisoned"); + 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); @@ -684,10 +683,12 @@ impl Rendition { } Poll::Pending }); - if startable.is_ready() { - run.startable = true; - } - startable + let Poll::Ready(resets) = startable else { + return Poll::Pending; + }; + // Tagged with the scanned window, so a clear racing this store leaves it stale. + self.run.lock().expect("run lock poisoned").startable = Some(resets); + Poll::Ready(()) } /// This rendition's DASH representation: its master-level metadata plus its track's diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index 6fbf437a77..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. @@ -214,18 +216,18 @@ impl Producer { rows: VecDeque::new(), sequence: 0, ended: false, + resets: 0, }), } } /// Append a row, evicting the front of the window past `window` (see [`evicts`]). With no - /// `window`, only source timeline pops remove rows. Returns whether the timeline jumped - /// backwards, which resets the window. - pub fn push(&self, row: Row, window: Option) -> bool { + /// `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 false; + return u64::MAX; }; - let mut restarted = false; if let Some(back) = state.rows.back() { // A backward jump in pts or segment number means the publisher restarted its timeline; @@ -233,7 +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(); - restarted = true; + state.resets += 1; } } if state.rows.is_empty() { @@ -254,7 +256,7 @@ impl Producer { state.rows.pop_front(); } state.sequence = state.rows.front().unwrap().segment; - restarted + state.resets } /// Remove rows whose source timeline indices fall within `range`. @@ -280,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) { @@ -372,11 +381,15 @@ impl Producer { } } - /// Poll `f` over the listed rows, waking on the next window change while it is pending. 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)) { - Poll::Ready(Ok(())) => Poll::Ready(()), + /// 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, } } @@ -639,16 +652,16 @@ mod tests { fn backwards_jump_resets_the_window() { let live = Producer::new(); let window = Some(Duration::from_secs(30)); - assert!(!live.push(row(0, 0, 10_000, 2_000), window)); - assert!(!live.push(row(1, 1, 12_000, 2_000), window)); - assert!(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. - assert!(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); } From df1addb55ca382074a7370a865583c6057a4cf74 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 18:45:51 -0700 Subject: [PATCH 9/9] fix(hls): a delayed latch never replaces a newer window's Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/rendition.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 26927da1b8..b95302a125 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -416,7 +416,7 @@ impl Rendition { let starts = self.is(reference) && row.starts_sync(); let resets = self.live.push(row, window); if starts { - self.run.lock().expect("run lock poisoned").startable = Some(resets); + self.latch(resets); } self.trim(); } @@ -687,10 +687,17 @@ impl Rendition { return Poll::Pending; }; // Tagged with the scanned window, so a clear racing this store leaves it stale. - self.run.lock().expect("run lock poisoned").startable = Some(resets); + 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