From 8b7577ccf6ba186e2cbfc53573a5317571644a59 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 09:38:31 -0700 Subject: [PATCH 1/2] quest: claim quest/m1/archive/hls-window Co-Authored-By: Claude Opus 5.5 From 821c89fba40f308b7343d06929593dc454defa44 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:03:45 -0700 Subject: [PATCH 2/2] feat(moq-hls): list a durable archive timeline without the live window A catalog archive entry with a store and no replay path declares its ranges durable on that broadcast, so the exporter lists everything the timeline retains and only its pops trim the playlists. DASH timeShiftBufferDepth becomes the listed span; --window still bounds live playlists and caps segment max-age. Co-Authored-By: Claude Opus 5.5 --- doc/bin/hls.md | 13 ++++-- quest/m1/archive/README.md | 11 +++-- quest/m1/archive/hls-window.md | 20 -------- rs/moq-cli/src/hls.rs | 1 + rs/moq-hls/src/export/archive_tests.rs | 65 +++++++++++++++++++++----- rs/moq-hls/src/export/mod.rs | 15 ++++++ rs/moq-hls/src/export/mpd.rs | 42 +++++++++++------ rs/moq-hls/src/export/rendition.rs | 5 +- rs/moq-hls/src/export/renditions.rs | 31 ++++++++---- rs/moq-hls/src/export/segments.rs | 27 ++++++----- 10 files changed, 154 insertions(+), 76 deletions(-) delete mode 100644 quest/m1/archive/hls-window.md diff --git a/doc/bin/hls.md b/doc/bin/hls.md index ddb9028029..cc85030f74 100644 --- a/doc/bin/hls.md +++ b/doc/bin/hls.md @@ -44,10 +44,12 @@ from the replayed timeline alone, and a segment GETs exactly one stored object of its rendition, 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. Set `--window` -to cover the recording. The playlist ends with `EXT-X-ENDLIST` only once the -reader's caller declares the recording finished; the store holds no completion -marker. +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 ranges 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 +declares the recording finished; the store holds no completion marker. 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 @@ -55,7 +57,8 @@ new one. An embedder of the library can also label the publisher's run with (`seg/{generation}.{segment}.m4s`), since a restarted publisher reuses segment numbers for different media. -`--window` sets the playlist duration (default 16 s), +`--window` sets the live 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. H.264/H.265 and AAC/Opus renditions are served. Import handles classic HLS; diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index fabafaf0e1..b9117707b2 100644 --- a/quest/m1/archive/README.md +++ b/quest/m1/archive/README.md @@ -38,8 +38,14 @@ The segment engine is in `rs/moq-mux/src/timeline.rs`: with no archive-specific code (`rs/moq-hls/src/export/archive_tests.rs`): playlists read only the timeline (an inline parameter set also GETs one keyframe group to build its init), and a segment GETs one object of its - rendition. The caller supplies the catalog, and `--window` must cover the - recording. + rendition. The caller supplies the catalog. +- A catalog `archive` entry with a `store` and no `replay` path declares its + ranges 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`). DASH `timeShiftBufferDepth` becomes the + listed span, and `--window` still bounds live playlists and caps segment + `max-age`. The catalog already states durability, so no per-broadcast + option or separate server is needed. `rs/moq-archive` stores the versioned objects on any `object_store::ObjectStore`: percent-encoded track names, `.info` JSON, the binary envelope, and put/get/list/delete. @@ -120,7 +126,6 @@ owned by that prerequisite, not duplicated in archive storage. ## Quests -- [Archive HLS window](/quest/m1/archive/hls-window.md) - serve a replayed recording's whole retained timeline without a server-wide `--window` - [Browser archive](/quest/m1/archive/browser.md) - the same contract for browser-published broadcasts - [DVR rewind](/quest/m1/archive/dvr.md) - seek through a bounded archive and return to live playback - [Archive proof](/quest/m1/archive/proof.md) - prove persistence ordering, selective reads, exact FETCH replay, and timeline-only HLS generation diff --git a/quest/m1/archive/hls-window.md b/quest/m1/archive/hls-window.md deleted file mode 100644 index 33cafc2466..0000000000 --- a/quest/m1/archive/hls-window.md +++ /dev/null @@ -1,20 +0,0 @@ -# [S] Archive HLS window - -## Goal - -`moq-hls` lists a replayed recording's whole retained timeline in HLS and DASH -without a server-wide `--window`, while live broadcasts keep a window within -the relay's cache. - -## Plan - -The export window (`moq_hls::export::Config::window`) trims every broadcast a -server exports. A recording served through `moq_archive::Reader` already -bounds itself: DVR expiry pops records from the replayed timeline, and an -unbounded archive never pops. Today an operator must raise `--window` past the -recording's length, which also inflates segment `Cache-Control: max-age` and -DASH `timeShiftBufferDepth` for every live broadcast on the same server. - -Decide how an export learns that a broadcast's timeline is authoritative (the -catalog's `archive` entry carrying a store or replay path, a per-broadcast -option, or a separate server) and cap the derived HTTP and DASH values. diff --git a/rs/moq-cli/src/hls.rs b/rs/moq-cli/src/hls.rs index f09a193cde..6a50170c84 100644 --- a/rs/moq-cli/src/hls.rs +++ b/rs/moq-cli/src/hls.rs @@ -34,6 +34,7 @@ pub struct ExportArgs { /// 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. #[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 3d5450c584..139434de14 100644 --- a/rs/moq-hls/src/export/archive_tests.rs +++ b/rs/moq-hls/src/export/archive_tests.rs @@ -196,11 +196,19 @@ fn record(segment: u64, pts: u64, duration: u64, tracks: &[(&str, u64, u64)]) -> record } -/// The catalog an exporter is handed: every rendition's config, plus the archive entry naming -/// the recording's timeline. Out-of-band configs, so no init needs media. -fn catalog() -> hang::Catalog { +/// The archive entry a replay advertises: the recording's timeline, durable in its store. +fn durable() -> hang::catalog::Archive { + let mut archive = hang::catalog::Archive::new(TIMELINE); + archive.store = Some("memory:///rec/".parse().unwrap()); + archive.version = Some(hang::catalog::Archive::VERSION); + archive +} + +/// The catalog an exporter is handed: every rendition's config, plus `archive`. Out-of-band +/// configs, so no init needs media. +fn catalog(archive: hang::catalog::Archive) -> hang::Catalog { let mut catalog = hang::Catalog::default(); - catalog.archive = Some(hang::catalog::Archive::new(TIMELINE)); + catalog.archive = Some(archive); for (name, width, height) in [("360p", 640, 360), ("1080p", 1920, 1080)] { let mut config = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8); config.coded_width = Some(width); @@ -224,7 +232,7 @@ struct Replay { } impl Replay { - async fn open(recording: &Recording, cache: u64) -> Self { + async fn open(recording: &Recording, cache: u64, archive: hang::catalog::Archive) -> 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(); @@ -235,7 +243,7 @@ impl Replay { let mut json = moq_json::snapshot::Config::default(); json.delta_ratio = 0; let mut catalog = moq_json::snapshot::Producer::new(track, json); - catalog.update(&self::catalog()).unwrap(); + catalog.update(&self::catalog(archive)).unwrap(); let config = reader::Config::new(TIMELINE).with_cache(cache); let reader = moq_archive::Reader::open(recording.store.clone(), &broadcast, config) @@ -296,8 +304,13 @@ fn is_media(path: &str) -> bool { /// Three aligned 2s segments: one keyframe group per video rendition, four audio groups each. async fn three_segments() -> Recording { + segments(3).await +} + +/// `count` aligned 2s segments, laid out like [`three_segments`]. +async fn segments(count: u64) -> Recording { let mut recording = Recording::new(&["360p", "1080p", "audio"]).await; - for segment in 0..3u64 { + for segment in 0..count { let pts = segment * 2_000_000; for video in ["360p", "1080p"] { recording.media(video, &[(segment, &[pts, pts + 1_000_000])]).await; @@ -324,7 +337,7 @@ async fn three_segments() -> Recording { #[tokio::test] async fn playlists_read_only_the_timeline_and_segments_one_object() { let recording = three_segments().await; - let replay = Replay::open(&recording, 64 * 1024 * 1024).await; + let replay = Replay::open(&recording, 64 * 1024 * 1024, durable()).await; let master = replay.broadcaster.master_playlist(None); assert!(master.contains("video/360p/media.m3u8") && master.contains("video/1080p/media.m3u8")); @@ -396,7 +409,7 @@ async fn playlists_read_only_the_timeline_and_segments_one_object() { async fn a_bounded_cache_rereads_evicted_objects() { let recording = three_segments().await; // Too small for any object, so nothing stays cached. - let replay = Replay::open(&recording, 1).await; + let replay = Replay::open(&recording, 1, durable()).await; replay.playlist(Kind::Audio, "audio").await; recording.gets(); @@ -433,7 +446,7 @@ async fn missing_track_segments_are_gaps_and_time_jumps_are_discontinuities() { .commit(&record(2, 10_000, 2000, &[("360p", 2, 2), ("1080p", 1, 1)]), 0) .await; - let replay = Replay::open(&recording, 64 * 1024 * 1024).await; + let replay = Replay::open(&recording, 64 * 1024 * 1024, durable()).await; let high = replay.playlist(Kind::Video, "1080p").await; let expected = concat!( "#EXTINF:2.00000,\nseg/0.m4s\n", @@ -460,7 +473,7 @@ async fn missing_track_segments_are_gaps_and_time_jumps_are_discontinuities() { #[tokio::test] async fn a_growing_recording_ends_only_on_caller_finality() { let mut recording = three_segments().await; - let mut replay = Replay::open(&recording, 64 * 1024 * 1024).await; + let mut replay = Replay::open(&recording, 64 * 1024 * 1024, durable()).await; replay.playlist(Kind::Video, "360p").await; // A DVR commit: segment 3 arrives and segment 0 expires. @@ -491,3 +504,33 @@ async fn a_growing_recording_ends_only_on_caller_finality() { .await; assert!(playlist.contains("seg/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. +#[tokio::test] +async fn a_durable_timeline_lists_past_the_window() { + let recording = segments(12).await; + let replay = Replay::open(&recording, 64 * 1024 * 1024, durable()).await; + let playlist = replay + .playlist_until(Kind::Video, "360p", |playlist| playlist.contains("seg/11.m4s\n")) + .await; + assert!(playlist.contains("#EXT-X-MEDIA-SEQUENCE:0\n"), "{playlist}"); + assert!(playlist.contains("seg/0.m4s\n"), "{playlist}"); + + for (kind, name) in [(Kind::Video, "360p"), (Kind::Video, "1080p"), (Kind::Audio, "audio")] { + replay.rendition(kind, name).init().await.unwrap(); + } + let manifest = replay.broadcaster.manifest(None).expect("manifest renders"); + assert!(manifest.contains("timeShiftBufferDepth=\"PT24.000S\""), "{manifest}"); + + let mut elsewhere = durable(); + elsewhere.replay = Some(moq_net::path::RelativeOwned::new("./recording")); + for archive in [hang::catalog::Archive::new(TIMELINE), elsewhere] { + let live = Replay::open(&recording, 64 * 1024 * 1024, archive).await; + let playlist = live + .playlist_until(Kind::Video, "360p", |playlist| playlist.contains("seg/11.m4s\n")) + .await; + assert!(!playlist.contains("seg/0.m4s\n"), "{playlist}"); + } +} diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 8b3b424189..e32845e5e7 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -66,6 +66,10 @@ 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. + /// + /// 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. pub window: Duration, } @@ -347,6 +351,9 @@ async fn watch_catalog( // records out to every rendition. if !timeline_started && let Some(archive) = catalog.archive.clone() { timeline_started = true; + if durable(&archive) { + renditions.fanout().unbound(); + } let watcher = tokio::spawn(watch_timeline(broadcast.clone(), archive, renditions.fanout())); *timeline_watcher.lock().unwrap() = Some(watcher); } @@ -363,6 +370,14 @@ async fn watch_catalog( renditions.close(); } +/// 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. +fn durable(archive: &hang::catalog::Archive) -> bool { + archive.store.is_some() && archive.replay.is_none() +} + /// The broadcast's timeline watcher: read the single timeline track and fan each record out /// to every rendition's window. async fn watch_timeline( diff --git a/rs/moq-hls/src/export/mpd.rs b/rs/moq-hls/src/export/mpd.rs index bea274ff80..11a4c333ee 100644 --- a/rs/moq-hls/src/export/mpd.rs +++ b/rs/moq-hls/src/export/mpd.rs @@ -61,8 +61,9 @@ pub(crate) struct Manifest { pub availability_start: Option, /// When this render happened (`MPD@publishTime`, dynamic only). pub publish: SystemTime, - /// The playlist window (`MPD@timeShiftBufferDepth`, dynamic only). - pub window: Duration, + /// The playlist window (`MPD@timeShiftBufferDepth`, dynamic only), or `None` when the + /// timeline bounds itself and the depth is the span it lists. + pub window: Option, /// The broadcast ended: render a `static` presentation instead of a `dynamic` one. pub finished: bool, /// Video representations, in catalog order. @@ -105,6 +106,23 @@ fn frame_rate(rate: f64) -> Option { } } +/// `units` of `timescale` as a [`Duration`]. +fn duration(units: u64, timescale: u32) -> Duration { + Duration::from_nanos((u128::from(units) * 1_000_000_000 / u128::from(timescale.max(1))) as u64) +} + +/// The longest span any representation lists, oldest segment start to newest segment end. +fn listed_span<'a>(representations: impl Iterator) -> Duration { + representations + .filter_map(|rep| { + let (first, _) = rep.segments.first()?; + let (last, d) = rep.segments.last()?; + Some(duration(last + d - first, rep.timescale)) + }) + .max() + .unwrap_or_default() +} + /// The largest listed segment duration in whole seconds, for `MPD@maxSegmentDuration` (and the /// update cadence). Like HLS's target duration, derived from the segments when the publisher /// declared no bound. @@ -225,10 +243,7 @@ pub(crate) fn render_manifest(manifest: &Manifest, query: Option<&str>) -> Strin let duration = representations() .filter_map(|rep| { let (t, d) = rep.segments.last()?; - let timescale = rep.timescale.max(1) as u64; - Some(Duration::from_nanos( - ((t + d) as u128 * 1_000_000_000 / timescale as u128) as u64, - )) + Some(duration(t + d, rep.timescale)) }) .max() .unwrap_or_default(); @@ -242,14 +257,15 @@ pub(crate) fn render_manifest(manifest: &Manifest, query: Option<&str>) -> Strin // Reload cadence and live delay follow HLS conventions: players refresh about once // per segment and sit a few segments behind the live edge (bounded by the window). let update = Duration::from_secs(target); - let delay = Duration::from_secs(3 * target).min(manifest.window.max(update)); + let window = manifest.window.unwrap_or_else(|| listed_span(representations())); + let delay = Duration::from_secs(3 * target).min(window.max(update)); let _ = write!( out, " type=\"dynamic\" availabilityStartTime=\"{}\" publishTime=\"{}\" minimumUpdatePeriod=\"{}\" timeShiftBufferDepth=\"{}\" suggestedPresentationDelay=\"{}\"", humantime::format_rfc3339_millis(availability), humantime::format_rfc3339_millis(manifest.publish), xs_duration(update), - xs_duration(manifest.window), + xs_duration(window), xs_duration(delay), ); } @@ -315,7 +331,7 @@ mod tests { let manifest = Manifest { availability_start: Some(SystemTime::UNIX_EPOCH + Duration::from_millis(1_751_846_400_123)), publish: SystemTime::UNIX_EPOCH + Duration::from_millis(1_751_846_410_000), - window: Duration::from_secs(16), + window: Some(Duration::from_secs(16)), finished: false, video: vec![video(vec![(0, 2_000), (2_000, 2_000)], false)], audio: vec![audio(vec![(0, 2_000), (2_000, 2_000)], false)], @@ -355,7 +371,7 @@ mod tests { let manifest = Manifest { availability_start: None, publish: SystemTime::UNIX_EPOCH, - window: Duration::from_secs(16), + window: Some(Duration::from_secs(16)), finished: true, // The window starts mid-broadcast: presentation time stays anchored at pts 0 (no // presentationTimeOffset), so the duration spans the lead-in and a live session @@ -379,7 +395,7 @@ mod tests { let manifest = Manifest { availability_start: Some(SystemTime::UNIX_EPOCH), publish: SystemTime::UNIX_EPOCH, - window: Duration::from_secs(16), + window: Some(Duration::from_secs(16)), finished: false, video: vec![video(vec![(0, 2_000)], false)], audio: Vec::new(), @@ -397,7 +413,7 @@ mod tests { let manifest = Manifest { availability_start: Some(SystemTime::UNIX_EPOCH), publish: SystemTime::UNIX_EPOCH, - window: Duration::from_secs(16), + window: Some(Duration::from_secs(16)), finished: false, video: vec![rep], audio: Vec::new(), @@ -420,7 +436,7 @@ mod tests { let manifest = Manifest { availability_start: Some(SystemTime::UNIX_EPOCH), publish: SystemTime::UNIX_EPOCH, - window: Duration::from_secs(16), + window: Some(Duration::from_secs(16)), finished: false, video: vec![video(vec![(0, 2_000)], false)], audio: vec![audio(Vec::new(), false)], diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index f1f1a03691..abb534c590 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -343,8 +343,9 @@ impl Rendition { } /// Feed one timeline record into this rendition's window: its own ranges (empty when the - /// record carries none for it, a gap), timed by the record. - pub(crate) fn push(&self, index: u64, entry: &Entry, discontinuity: u64, window: Duration) { + /// record carries none for it, a gap), timed by the record. With no `window`, only source + /// timeline pops trim it. + pub(crate) fn push(&self, index: u64, entry: &Entry, discontinuity: u64, window: Option) { if !self.media.admits(&self.live) { return; } diff --git a/rs/moq-hls/src/export/renditions.rs b/rs/moq-hls/src/export/renditions.rs index 63815fed7c..ab66923500 100644 --- a/rs/moq-hls/src/export/renditions.rs +++ b/rs/moq-hls/src/export/renditions.rs @@ -65,6 +65,10 @@ struct Feed { anchor: Option, /// 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. + window: Option, } /// The producing side of a broadcast's rendition set. @@ -87,8 +91,6 @@ pub(crate) struct Producer { #[derive(Clone)] pub(crate) struct Fanout { feed: Arc>, - /// The playlist window duration (see [`Config::window`](super::Config::window)), applied on every push. - window: Duration, } impl Producer { @@ -104,8 +106,8 @@ impl Producer { closed: false, anchor: None, generation: None, + window: Some(window), })), - window, }, } } @@ -115,9 +117,10 @@ impl Producer { self.fanout.clone() } - /// The playlist window duration every rendition's window is trimmed to. - pub fn window(&self) -> Duration { - self.fanout.window + /// The playlist window duration every rendition's window is trimmed to, or `None` when + /// only the source timeline trims them. + pub fn window(&self) -> Option { + self.fanout.feed.lock().unwrap().window } /// The estimated wall-clock time of timeline `pts` 0 (see [`Feed::anchor`]); `None` until @@ -184,6 +187,12 @@ impl Producer { } impl Fanout { + /// List every record the source timeline retains, trimming only on its pops, instead of + /// a window. Call before the first record. + pub fn unbound(&self) { + self.feed.lock().unwrap().window = None; + } + /// Fan one timeline record out to every living rendition, and into the replay history. pub fn push(&self, index: u64, entry: Entry) { let mut feed = self.feed.lock().unwrap(); @@ -205,17 +214,19 @@ impl Fanout { let pts = Duration::from(entry.pts); let discontinuity = feed.discontinuities.stamp(pts, pts + entry.duration); feed.history.push_back((index, entry.clone(), discontinuity)); - while feed.history.len() >= 2 { + 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 < self.window { + if span < window { break; } feed.history.pop_front(); } - let window = self.window; feed.targets.retain(|target| { let Some(rendition) = target.upgrade() else { return false; @@ -335,7 +346,7 @@ impl Producer { let mut feed = self.fanout.feed.lock().unwrap(); rendition.label(feed.generation.clone()); for (index, entry, discontinuity) in &feed.history { - rendition.push(*index, entry, *discontinuity, self.fanout.window); + rendition.push(*index, entry, *discontinuity, feed.window); } if feed.ended { rendition.end(); diff --git a/rs/moq-hls/src/export/segments.rs b/rs/moq-hls/src/export/segments.rs index 015f1f2509..6b8a92a0b6 100644 --- a/rs/moq-hls/src/export/segments.rs +++ b/rs/moq-hls/src/export/segments.rs @@ -166,8 +166,9 @@ impl Producer { } } - /// Append a row, evicting the front of the window past `window`. - pub fn push(&self, row: Row, window: Duration) { + /// Append a row, evicting the front of the window past `window`. With no `window`, only + /// source timeline pops remove rows. + pub fn push(&self, row: Row, window: Option) { let Ok(mut state) = self.state.write() else { return; }; @@ -187,7 +188,9 @@ impl Producer { state.rows.push_back(row); // Evict from the front while the remaining rows still cover the window. - while state.rows.len() >= 2 { + 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 { break; @@ -439,8 +442,8 @@ mod tests { #[test] fn every_row_is_listed() { let live = Producer::new(); - live.push(row(0, 0, 0, 2_000), Duration::from_secs(30)); - live.push(row(1, 1, 2_000, 2_000), Duration::from_secs(30)); + live.push(row(0, 0, 0, 2_000), Some(Duration::from_secs(30))); + live.push(row(1, 1, 2_000, 2_000), Some(Duration::from_secs(30))); let window = live.window(); assert_eq!(window.sequence, 0); @@ -456,7 +459,7 @@ mod tests { #[test] fn window_evicts_and_advances_sequence() { let live = Producer::new(); - let window = Duration::from_secs(4); + let window = Some(Duration::from_secs(4)); for i in 0..6u64 { live.push(row(i, i, i * 2_000, 2_000), window); } @@ -473,7 +476,7 @@ mod tests { #[test] fn source_window_pop_removes_playlist_rows() { let live = Producer::new(); - let window = Duration::from_secs(30); + let window = Some(Duration::from_secs(30)); for i in 0..4u64 { let mut row = row(i, i, i * 2_000, 2_000); row.index = i + 10; @@ -497,9 +500,9 @@ mod tests { #[test] fn a_skipped_source_range_clears_rows_before_the_next_segment() { let live = Producer::new(); - live.push(row(4, 4, 8_000, 2_000), Duration::from_secs(10)); + live.push(row(4, 4, 8_000, 2_000), Some(Duration::from_secs(10))); live.clear(); - live.push(row(10, 10, 20_000, 2_000), Duration::from_secs(10)); + live.push(row(10, 10, 20_000, 2_000), Some(Duration::from_secs(10))); let snapshot = live.window(); assert_eq!(snapshot.sequence, 10); @@ -567,7 +570,7 @@ mod tests { #[test] fn segment_ranges_and_gaps() { let live = Producer::new(); - let window = Duration::from_secs(30); + let window = Some(Duration::from_secs(30)); live.push(row(0, 0, 0, 1_000), window); // Segment 1 is a gap for this rendition: no ranges. live.push( @@ -595,7 +598,7 @@ mod tests { #[test] fn backwards_jump_resets_the_window() { let live = Producer::new(); - let window = Duration::from_secs(30); + 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 @@ -612,7 +615,7 @@ mod tests { #[test] fn next_after_walks_segments() { let live = Producer::new(); - let window = Duration::from_secs(30); + let window = Some(Duration::from_secs(30)); live.push(row(0, 0, 0, 2_000), window); live.push(row(1, 1, 2_000, 2_000), window);