feat(moq-hls): list a durable archive timeline without the live window

doc/bin/hls.md

@@ -44,18 +44,21 @@ 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
 `Broadcaster::set_generation`. Every segment URL then carries it
 (`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;

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

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,
 

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}");
+	}
+}

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(

rs/moq-hls/src/export/mpd.rs

@@ -61,8 +61,9 @@ pub(crate) struct Manifest {
 	pub availability_start: Option<SystemTime>,
 	/// 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<Duration>,
 	/// 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<String> {
 	}
 }
 
+/// `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<Item = &'a Representation>) -> 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)],

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<Duration>) {
 		if !self.media.admits(&self.live) {
 			return;
 		}