feat(archive): rewind a track into its recording, then splice to live

doc/lib/rs/index.md

@@ -19,7 +19,7 @@ The reference implementation. Every crate is on
 | [moq-uring](https://docs.rs/moq-uring) | Experimental Linux io\_uring worker: one pinned thread per ring serving moq-lite over its own QUIC stack. |
 | [hang](/lib/rs/hang) | The media layer: catalog, containers, ordered frame delivery. |
 | [moq-mux](/lib/rs/moq-mux) | Import and export fMP4/CMAF, MPEG-TS, Matroska, FLV, and Annex-B. |
-| [moq-archive](https://docs.rs/moq-archive) | Versioned hang recordings on any `object_store` backend: track layout, `.info` JSON, and segment objects. |
+| [moq-archive](https://docs.rs/moq-archive) | Versioned hang recordings on any `object_store` backend: record, replay, and rewind a live broadcast into its recording. |
 | [moq-video](/lib/rs/moq-video) | Native capture, hardware encode/decode (Apple, Windows, NVIDIA, VAAPI, V4L2, Android), and GPU rendering. |
 | [moq-v4l](https://docs.rs/moq-v4l) | Safe Video4Linux 2 bindings with the kernel headers checked in, so a build needs no libclang. |
 | [moq-audio](/lib/rs/moq-audio) | Microphone and speaker, Opus/PCM/AAC codecs, echo cancellation. |

quest/m1/archive/README.md

@@ -60,6 +60,13 @@ unreferenced group objects one grace period after recovery.
 supplied `broadcast::Producer` and serves FETCH through `track::Dynamic` with a byte-bounded
 object LRU. `Reader::refresh` follows by listing timeline keys after its cursor, so gaps and
 DVR expiry recover from the next checkpoint; `Reader::finish` applies out-of-band finality.
+`moq_archive::Rewind` (`rs/moq-archive/src/rewind/mod.rs`) seeks a viewer's track into the
+replayed timeline, FETCHes each advertised group in order from the replay broadcast, and
+subscribes to the live broadcast from the first group past the recording's head. The
+timeline decides what is requested: expired, unadvertised, or not-found groups are gaps.
+A served group that later expires stays in the replay track's `moq_net` cache until the
+pool reclaims it; no group-eviction API is needed, because no viewer following the
+timeline asks for it and its bytes are immutable under never-reused sequences.
 
 ### Format
 
@@ -127,7 +134,6 @@ owned by that prerequisite, not duplicated in archive storage.
 ## Quests
 
 - [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
 
 ## Related

rs/moq-archive/README.md

@@ -14,6 +14,10 @@ that needs runtime dispatch passes `Arc<dyn ObjectStore>`. A broadcast
 advertises its recording through the catalog's
 [`archive`](https://doc.moq.dev/concept/hang) entry.
 
+`Writer` records selected tracks of a broadcast, `Reader` serves a recording
+back through a broadcast, and `Rewind` seeks a viewer into a recording before
+splicing it back onto the live broadcast.
+
 ```bash
 cargo add moq-archive
 ```

rs/moq-archive/src/reader/index.rs → rs/moq-archive/src/index.rs

@@ -1,22 +1,24 @@
 use std::collections::{BTreeMap, HashMap};
-use std::ops::{Range, RangeInclusive};
+use std::ops::{Bound, Range, RangeInclusive};
 
 use hang::timeline::Record;
 
 use crate::path::check_range;
 
 /// One track's stored object for one record: its filename bounds and the exact runs it holds.
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub(super) struct Span {
+pub(crate) struct Span {
 	/// Inclusive first-to-last group sequence, the object's filename bounds.
 	pub bounds: RangeInclusive<u64>,
 	/// The advertised runs, ascending and nonoverlapping; groups between them never existed.
 	pub runs: Vec<RangeInclusive<u64>>,
+	/// The advertising record's start, in timeline units.
+	pub pts: u64,
 }
 
 impl Span {
 	/// Build from a record's ranges, refusing empty, reversed, out-of-range, or unordered runs.
-	fn new(ranges: &[hang::timeline::Range]) -> Option<Self> {
+	fn new(ranges: &[hang::timeline::Range], pts: u64) -> Option<Self> {
 		let mut runs: Vec<RangeInclusive<u64>> = Vec::with_capacity(ranges.len());
 		for range in ranges {
 			let run = range.start..=range.end;
@@ -29,7 +31,7 @@ impl Span {
 			runs.push(run);
 		}
 		let bounds = *runs.first()?.start()..=*runs.last()?.end();
-		Some(Self { bounds, runs })
+		Some(Self { bounds, runs, pts })
 	}
 
 	fn contains(&self, group: u64) -> bool {
@@ -39,7 +41,7 @@ impl Span {
 
 /// The committed group ranges advertised by the replayed timeline window.
 #[derive(Default)]
-pub(super) struct Index {
+pub(crate) struct Index {
 	/// Per track, each span keyed by its smallest group sequence.
 	tracks: HashMap<String, BTreeMap<u64, Span>>,
 	/// Per window index, the `(track, smallest)` spans that record added, so a pop can evict them.
@@ -54,7 +56,7 @@ impl Index {
 	pub fn push(&mut self, index: u64, record: &Record) {
 		let mut added = Vec::new();
 		for (track, ranges) in &record.tracks {
-			let Some(span) = Span::new(ranges) else {
+			let Some(span) = Span::new(ranges, record.pts) else {
 				tracing::warn!(track, segment = record.segment, "ignoring malformed archive ranges");
 				continue;
 			};
@@ -93,6 +95,32 @@ impl Index {
 		self.tracks.contains_key(track)
 	}
 
+	/// The first group at or after `group` that the retained window commits on `track`.
+	pub fn next(&self, track: &str, group: u64) -> Option<u64> {
+		let spans = self.tracks.get(track)?;
+		let within = spans.range(..=group).next_back().and_then(|(_, span)| {
+			let run = span.runs.iter().find(|run| *run.end() >= group)?;
+			Some(group.max(*run.start()))
+		});
+		within.or_else(|| {
+			let (smallest, _) = spans.range((Bound::Excluded(group), Bound::Unbounded)).next()?;
+			Some(*smallest)
+		})
+	}
+
+	/// The first group of the latest `track` span whose record starts at or before `pts`, or of
+	/// the earliest span when every record starts later.
+	pub fn seek(&self, track: &str, pts: u64) -> Option<u64> {
+		let spans = self.tracks.get(track)?;
+		let earliest = spans.values().next()?;
+		let span = spans
+			.values()
+			.take_while(|span| span.pts <= pts)
+			.last()
+			.unwrap_or(earliest);
+		Some(*span.bounds.start())
+	}
+
 	/// The span advertising `group` on `track`, if the retained window commits it.
 	pub fn get(&self, track: &str, group: u64) -> Option<Span> {
 		let (_, span) = self.tracks.get(track)?.range(..=group).next_back()?;
@@ -133,6 +161,50 @@ mod tests {
 		assert!(!index.has_track("chat"));
 	}
 
+	#[test]
+	fn next_skips_gaps_and_popped_records() {
+		let mut index = Index::default();
+		index.push(0, &record(0, &[("audio", &[(0, 2), (5, 6)])]));
+		index.push(1, &record(1, &[("audio", &[(9, 9)]), ("video", &[(3, 3)])]));
+
+		assert_eq!(index.next("audio", 0), Some(0));
+		assert_eq!(index.next("audio", 2), Some(2));
+		assert_eq!(index.next("audio", 3), Some(5), "internal gap");
+		assert_eq!(index.next("audio", 7), Some(9), "gap between records");
+		assert_eq!(index.next("audio", 10), None, "past the newest record");
+		assert_eq!(index.next("video", 0), Some(3));
+		assert_eq!(index.next("chat", 0), None);
+
+		index.pop(0..1);
+		assert_eq!(
+			index.next("audio", 1),
+			Some(9),
+			"expired groups skip to the retained window"
+		);
+	}
+
+	#[test]
+	fn seek_picks_the_segment_starting_at_or_before() {
+		let mut index = Index::default();
+		// Records start at segment * 1000.
+		index.push(0, &record(0, &[("video", &[(0, 1)])]));
+		index.push(1, &record(1, &[("audio", &[(0, 0)])]));
+		index.push(2, &record(2, &[("video", &[(4, 5)])]));
+
+		assert_eq!(index.seek("video", 0), Some(0));
+		assert_eq!(index.seek("video", 1999), Some(0), "segment 1 has no video");
+		assert_eq!(index.seek("video", 2000), Some(4));
+		assert_eq!(index.seek("video", u64::MAX), Some(4));
+		assert_eq!(index.seek("chat", 0), None);
+
+		index.pop(0..1);
+		assert_eq!(
+			index.seek("video", 0),
+			Some(4),
+			"before the window starts at its oldest span"
+		);
+	}
+
 	#[test]
 	fn pop_evicts_only_the_popped_records() {
 		let mut index = Index::default();

rs/moq-archive/src/lib.rs

@@ -2,8 +2,8 @@
 //!
 //! The crate owns the portable layout and codecs: percent-encoded track names, `.info` JSON,
 //! the binary segment envelope, and put/get/list/delete. A [`Writer`] records selected tracks of
-//! a broadcast into those objects, and a [`Reader`] serves a recording back through a `moq_net`
-//! broadcast. Callers that need runtime dispatch
+//! a broadcast into those objects, a [`Reader`] serves a recording back through a `moq_net`
+//! broadcast, and a [`Rewind`] seeks a viewer back into it before returning to live. Callers that need runtime dispatch
 //! supply `Arc<dyn ObjectStore>`; the archive API itself stays generic.
 //!
 //! Group bounds are finite inclusive ranges in first-to-last order:
@@ -16,10 +16,12 @@
 pub use object_store;
 
 mod error;
+mod index;
 pub mod info;
 mod path;
 pub mod reader;
 mod recover;
+pub mod rewind;
 pub mod segment;
 pub mod store;
 pub mod writer;
@@ -28,6 +30,7 @@ pub use error::{Error, Result};
 pub use info::Info;
 pub use path::Key;
 pub use reader::Reader;
+pub use rewind::Rewind;
 pub use segment::{Frame, Group, Object};
 pub use store::Store;
 pub use writer::Writer;

rs/moq-archive/src/reader/mod.rs

@@ -21,8 +21,6 @@
 //! # }
 //! ```
 
-mod index;
-
 use std::future::Future;
 use std::ops::RangeInclusive;
 use std::sync::{Arc, Mutex};
@@ -35,7 +33,7 @@ use moq_json::window;
 use moq_net::{Timescale, Timestamp, broadcast, group, track};
 use object_store::ObjectStore;
 
-use self::index::{Index, Span};
+use crate::index::{Index, Span};
 use crate::store::list::Query;
 use crate::{Error, Key, Object, Result, Store};