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);