feat(archive): resume a recording on restart

quest/m1/archive/README.md

@@ -40,7 +40,10 @@ percent-encoded track names, `.info` JSON, the binary envelope, and put/get/list
 `moq_archive::Writer` (`rs/moq-archive/src/writer.rs`) records enrolled tracks
 through `Deferred`, omits failed tracks with `Pending::omit`, stores each
 segment's timeline groups after `Producer::flush`, and expires DVR segments
-with a deletion grace. It refuses a prefix that already holds a timeline.
+with a deletion grace. On a prefix that already holds a recording, it replays the
+retained timeline from a checkpoint through `timeline::Producer::resume`, refuses
+groups at or below each track's largest stored group, and a DVR deletes
+unreferenced group objects one grace period after recovery.
 `moq_archive::Reader` (`rs/moq-archive/src/reader/mod.rs`) replays the timeline onto a
 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
@@ -113,7 +116,6 @@ owned by that prerequisite, not duplicated in archive storage.
 
 - [Browser archive](/quest/m1/archive/browser.md) - the same contract for browser-published broadcasts
 - [Offline archive HLS](/quest/m1/archive/hls.md) - render playlists from the archive timeline and fetch segment media lazily
-- [Resume a recording](/quest/m1/archive/recovery.md) - recover the retained timeline on restart and clean up DVR orphans
 - [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
 

quest/m1/archive/dvr.md

@@ -7,10 +7,9 @@ same timeline and group-range objects as an unbounded archive.
 
 ## Plan
 
-The recording writer owns retention and deletion grace, and
-[Resume a recording](/quest/m1/archive/recovery.md) owns checkpoint recovery
-and restart cleanup. This quest consumes that contract and owns viewer
-seek and return-to-live behavior, not a second writer implementation.
+The recording writer owns retention, deletion grace, checkpoint recovery, and
+restart cleanup. This quest consumes that contract and owns viewer seek and
+return-to-live behavior, not a second writer implementation.
 
 The player reads the archive timeline, FETCHes old groups through the normal
 miss chain, and splices back to SUBSCRIBE at the live edge without opening a
@@ -27,10 +26,6 @@ The reader evicts popped spans from its object cache, but a group it already
 served stays in `moq_net`'s track cache until the pool reclaims it. Decide
 whether expiry during a seek needs a group eviction API in `moq-net`.
 
-## Required
-
-- [Resume a recording](/quest/m1/archive/recovery.md)
-
 ## Closes
 
 - [#2275](https://github.com/moq-dev/moq/issues/2275) - close this issue when the quest finishes

quest/m1/archive/proof.md

@@ -63,5 +63,4 @@ object directly without any listing or separate index object.
 
 ## Required
 
-- [Resume a recording](/quest/m1/archive/recovery.md)
 - [Offline archive HLS](/quest/m1/archive/hls.md)

rs/moq-archive/src/error.rs

@@ -78,10 +78,6 @@ pub enum Error {
 	#[error("moq: {0}")]
 	Moq(String),
 
-	/// The prefix already holds a recording timeline; resuming is unsupported.
-	#[error("prefix already holds a recording: {0}")]
-	Occupied(String),
-
 	/// The track was already enrolled, or is the recording's own timeline.
 	#[error("track already enrolled: {0}")]
 	Enrolled(String),
@@ -90,7 +86,7 @@ pub enum Error {
 	#[error("source: {0}")]
 	Source(String),
 
-	/// The recording's timeline could not be segmented or published.
+	/// The recording's timeline could not be recovered, segmented, or published.
 	#[error("timeline: {0}")]
 	Timeline(String),
 

rs/moq-archive/src/lib.rs

@@ -19,6 +19,7 @@ mod error;
 pub mod info;
 mod path;
 pub mod reader;
+mod recover;
 pub mod segment;
 pub mod store;
 pub mod writer;

rs/moq-archive/src/recover.rs

@@ -0,0 +1,185 @@
+//! Recover a recording so a restarted [`Writer`](crate::Writer) continues it.
+
+use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
+use std::ops::RangeInclusive;
+
+use futures::TryStreamExt;
+use hang::timeline::Record;
+use moq_json::window::{self, Checkpoint};
+use object_store::ObjectStore;
+
+use crate::store::list::Query;
+use crate::{Error, Key, Object, Result, Store};
+
+/// What a restarted writer continues from.
+pub(crate) struct Recovery {
+	/// The retained timeline window, or `None` when no timeline object exists.
+	pub checkpoint: Option<Checkpoint<Record>>,
+	/// The next stored timeline group sequence, so the stored numbering keeps increasing.
+	pub sequence: u64,
+	/// Per track, the largest stored group; new groups must exceed it so no object overlaps.
+	pub floors: HashMap<String, u64>,
+	/// Stored group objects no retained record references. Only collected for a DVR.
+	pub orphans: Vec<Key>,
+}
+
+/// List the whole recording, then replay its timeline from a retained checkpoint.
+///
+/// A DVR (`complete`) replays far enough back to recover every retained record and reports the
+/// unreferenced group objects; an archive only needs the newest checkpoint. Any listing, GET, or
+/// decode failure fails recovery, so a caller never acts on a partial view.
+pub(crate) async fn recover<S: ObjectStore>(store: &Store<S>, timeline: &str, complete: bool) -> Result<Recovery> {
+	let entries: Vec<_> = store.list(&Query::new()).try_collect().await?;
+	let mut segments = Vec::new();
+	let mut groups = Vec::new();
+	let mut floors = HashMap::new();
+	for entry in entries {
+		match entry.key {
+			Key::Segments { track, segment } if track == timeline => segments.push(segment),
+			Key::Groups { track, range } if track != timeline => {
+				raise(&mut floors, &track, *range.end());
+				groups.push(Key::Groups { track, range });
+			}
+			_ => {}
+		}
+	}
+	segments.sort_unstable();
+
+	let (checkpoint, sequence) = match (segments.first(), segments.last()) {
+		(Some(&first), Some(&last)) => {
+			if segments.len() as u64 != last - first + 1 {
+				return Err(Error::Timeline(format!(
+					"timeline segments {first}..={last} are not contiguous"
+				)));
+			}
+			let (checkpoint, sequence) = replay(store, timeline, first..=last, complete).await?;
+			// The newest object is segment `last`, so the window must end on the next one.
+			// A shorter window would resume onto that segment and collide with it.
+			let next = last.checked_add(1).ok_or(Error::Overflow)?;
+			if checkpoint.range.end != next {
+				return Err(Error::Timeline(format!(
+					"recovered window ends at {}, not segment {next}",
+					checkpoint.range.end
+				)));
+			}
+			(Some(checkpoint), sequence)
+		}
+		_ => (None, 0),
+	};
+
+	let mut referenced = HashSet::new();
+	for record in checkpoint.iter().flat_map(|checkpoint| &checkpoint.records) {
+		for (track, ranges) in &record.tracks {
+			if let (Some(first), Some(last)) = (ranges.first(), ranges.last()) {
+				raise(&mut floors, track, last.end);
+				referenced.insert(Key::groups(track.clone(), first.start..=last.end)?);
+			}
+		}
+	}
+	let orphans = match complete {
+		true => groups.into_iter().filter(|key| !referenced.contains(key)).collect(),
+		false => Vec::new(),
+	};
+
+	Ok(Recovery {
+		checkpoint,
+		sequence,
+		floors,
+		orphans,
+	})
+}
+
+fn raise(floors: &mut HashMap<String, u64>, track: &str, group: u64) {
+	let floor = floors.entry(track.to_string()).or_insert(group);
+	*floor = (*floor).max(group);
+}
+
+/// Replay `segments` from the newest checkpoint that restates every record `complete` needs.
+///
+/// Returns the retained window and the next timeline group sequence.
+async fn replay<S: ObjectStore>(
+	store: &Store<S>,
+	timeline: &str,
+	segments: RangeInclusive<u64>,
+	complete: bool,
+) -> Result<(Checkpoint<Record>, u64)> {
+	// Every object opens with a checkpoint. The newest one's offset bounds what is still retained,
+	// so walk back until a checkpoint restates from there.
+	let mut objects = VecDeque::new();
+	let mut needed = None;
+	for segment in segments.rev() {
+		let object = store.get_segments(timeline, segment).await?;
+		let (offset, start) = checkpoint(&object)?;
+		let needed = *needed.get_or_insert(offset);
+		objects.push_front(object);
+		if !complete || start <= needed {
+			break;
+		}
+	}
+
+	let mut decoder = decoder();
+	let mut records = BTreeMap::new();
+	for object in &objects {
+		for stored in &object.groups {
+			let mut group = decoder.group();
+			for frame in &stored.frames {
+				group.decode(&frame.payload).map_err(json_error)?;
+			}
+		}
+		while let Some(event) = decoder.next_event() {
+			match event {
+				window::Event::Push { index, value } => {
+					records.insert(index, value);
+				}
+				window::Event::Pop(range) | window::Event::Skip(range) => {
+					let tail = records.split_off(&range.end);
+					records.retain(|index, _| *index < range.start);
+					records.extend(tail);
+				}
+				_ => {}
+			}
+		}
+	}
+
+	// The records must be a contiguous suffix of the window, and all of it for a DVR.
+	let range = decoder.range();
+	let start = records.keys().next().copied().unwrap_or(range.end);
+	if records.len() as u64 != range.end - start || (complete && start != range.start) {
+		return Err(Error::Timeline(format!(
+			"cannot recover timeline window {range:?} from the retained checkpoints"
+		)));
+	}
+
+	let last = objects.back().and_then(|object| object.groups.last());
+	let sequence = last.map_or(Ok(0), |group| group.sequence.checked_add(1).ok_or(Error::Overflow))?;
+	let checkpoint = Checkpoint {
+		range,
+		records: records.into_values().collect(),
+	};
+	Ok((checkpoint, sequence))
+}
+
+/// The retained offset and first restated index of the checkpoint opening `object`.
+fn checkpoint(object: &Object) -> Result<(u64, u64)> {
+	let frame = object
+		.groups
+		.first()
+		.and_then(|group| group.frames.first())
+		.ok_or_else(|| Error::Timeline("timeline object has no checkpoint".into()))?;
+	let mut decoder = decoder();
+	decoder.group().decode(&frame.payload).map_err(json_error)?;
+	let offset = decoder.range().start;
+	let start = match decoder.next_event() {
+		Some(window::Event::Skip(skipped)) => skipped.end,
+		_ => offset,
+	};
+	Ok((offset, start))
+}
+
+fn decoder() -> window::Decoder<Record> {
+	window::Decoder::new(window::ConsumerConfig::default().with_compression(true))
+}
+
+fn json_error(err: moq_json::Error) -> Error {
+	Error::Timeline(err.to_string())
+}