diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index e4cdd70249..f16037033a 100644 --- a/quest/m1/archive/README.md +++ b/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 diff --git a/quest/m1/archive/dvr.md b/quest/m1/archive/dvr.md index 685185fa65..096633a809 100644 --- a/quest/m1/archive/dvr.md +++ b/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 diff --git a/quest/m1/archive/proof.md b/quest/m1/archive/proof.md index c28c05284b..b8cb7e0f47 100644 --- a/quest/m1/archive/proof.md +++ b/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) diff --git a/quest/m1/archive/recovery.md b/quest/m1/archive/recovery.md deleted file mode 100644 index 70b302abd5..0000000000 --- a/quest/m1/archive/recovery.md +++ /dev/null @@ -1,26 +0,0 @@ -# [M] Resume a recording - -## Goal - -A writer restarting on a prefix it exclusively owns recovers the retained -timeline and continues the same recording instead of refusing the prefix. - -## Plan - -`moq_archive::Writer::new` refuses any prefix that already holds a timeline. -Replace that refusal with recovery: list and replay the timeline's -`segments/` objects from a retained checkpoint, then resume the timeline -producer at the next segment ID with the recovered window. That needs a way -to seed `moq_mux::timeline::Producer` with a window and next segment, which it -lacks today. - -For a DVR, reconcile a complete `groups/` listing of every recorded track -against the recovered records before accepting input. Wait the deletion grace -period from successful recovery, then delete unreferenced group objects left -by interrupted expiration or uploads. Failed or incomplete recovery or listing -deletes nothing. Preserve `.info` and timeline checkpoint objects. - -An unbounded archive resumes the same way without deleting anything. Source -group sequences that restart below the recovered ranges must not produce -overlapping objects; decide whether that refuses the track or starts a new -recording. diff --git a/rs/moq-archive/src/error.rs b/rs/moq-archive/src/error.rs index 85010486b9..a8ef749d83 100644 --- a/rs/moq-archive/src/error.rs +++ b/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), diff --git a/rs/moq-archive/src/lib.rs b/rs/moq-archive/src/lib.rs index b582fffbbd..3d5d9bb1d8 100644 --- a/rs/moq-archive/src/lib.rs +++ b/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; diff --git a/rs/moq-archive/src/recover.rs b/rs/moq-archive/src/recover.rs new file mode 100644 index 0000000000..f7ccfd5ace --- /dev/null +++ b/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>, + /// 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, + /// Stored group objects no retained record references. Only collected for a DVR. + pub orphans: Vec, +} + +/// 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(store: &Store, timeline: &str, complete: bool) -> Result { + 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, 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( + store: &Store, + timeline: &str, + segments: RangeInclusive, + complete: bool, +) -> Result<(Checkpoint, 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 { + window::Decoder::new(window::ConsumerConfig::default().with_compression(true)) +} + +fn json_error(err: moq_json::Error) -> Error { + Error::Timeline(err.to_string()) +} diff --git a/rs/moq-archive/src/writer.rs b/rs/moq-archive/src/writer.rs index 766794a75e..5f05cdc083 100644 --- a/rs/moq-archive/src/writer.rs +++ b/rs/moq-archive/src/writer.rs @@ -6,6 +6,8 @@ //! failed, commits the record through its own timeline encoder, and stores that segment's timeline //! groups before starting the next one. The timeline therefore only advertises durable objects. //! +//! A writer started on a prefix that already holds a recording resumes it: see [`Writer::new`]. +//! //! ```no_run //! # async fn example(source: moq_net::broadcast::Consumer) -> moq_archive::Result<()> { //! use moq_archive::object_store::memory::InMemory; @@ -37,8 +39,8 @@ use object_store::ObjectStore; use tokio::sync::{mpsc, watch}; use tokio::time::Instant; +use crate::recover::recover; use crate::segment::{Frame, Group, Object}; -use crate::store::list::Query; use crate::{Error, Info, Key, Result, Store}; /// Subscribers ask for every cached group; the publisher clamps this to its own max age. @@ -95,6 +97,8 @@ pub struct Writer { commands: mpsc::UnboundedReceiver, committer: Committer, grace: Option, + /// Deadlines for deleting expired or orphaned objects, oldest first. + deletions: VecDeque<(Instant, Vec)>, // Owns the recording's timeline track. _timeline: broadcast::Producer, } @@ -128,6 +132,8 @@ struct Shared { timeline: String, /// Every name ever enrolled. A name is never reused, so its object ranges stay increasing. enrolled: Mutex>, + /// Per track, the largest group a resumed recording already stored. + floors: HashMap, } enum Command { @@ -143,20 +149,27 @@ enum Command { } impl Writer { - /// Start a recording under `store`'s prefix, reading tracks from `source`. + /// Start a recording under `store`'s prefix, reading tracks from `source`, or resume the one + /// already there. /// - /// Refuses a prefix that already holds a timeline: resuming a recording is unsupported. + /// Resuming replays the retained timeline and continues at the next segment. A track refuses + /// any group at or below the largest one stored for it, so a source whose group sequences + /// restarted needs a new prefix. A DVR also deletes, one grace period after recovery, every + /// group object its retained records do not reference, such as interrupted expirations and + /// uploads. The writer must own the prefix exclusively. Fails, deleting nothing, when the + /// recording cannot be listed or its timeline cannot be replayed. pub async fn new(store: Store, source: broadcast::Consumer, config: Config) -> Result { + let section = timeline::Segmenter::new(config.timeline.clone()).section(); + let recovery = recover(&store, §ion.track, config.retention.is_some()).await?; + let broadcast = broadcast::Info::new().produce(); - let timeline = timeline::Producer::new(&broadcast, config.timeline); + let timeline = match &recovery.checkpoint { + Some(checkpoint) => { + timeline::Producer::resume(&broadcast, config.timeline, checkpoint).map_err(timeline_error)? + } + None => timeline::Producer::new(&broadcast, config.timeline), + }; let deferred = timeline.deferred().map_err(timeline_error)?; - let section = timeline.section(); - - let mut existing = store.list(&Query::segments(§ion.track)?); - if let Some(entry) = existing.next().await { - return Err(Error::Occupied(store.path(&entry?.key)?.to_string())); - } - drop(existing); let replay = track::Subscription::default().with_max_age(REPLAY); let groups = broadcast @@ -175,9 +188,16 @@ impl Writer { source, timeline: section.track, enrolled: Mutex::new(HashSet::new()), + floors: recovery.floors, }); let (commands, receiver) = mpsc::unbounded_channel(); let retention = config.retention; + let mut deletions = VecDeque::new(); + if let Some(retention) = &retention + && !recovery.orphans.is_empty() + { + deletions.push_back((Instant::now() + retention.grace, recovery.orphans)); + } Ok(Self { control: Control { shared: shared.clone(), @@ -191,9 +211,11 @@ impl Writer { groups, timescale: section.timescale.into(), retention: retention.as_ref().map(|r| r.window), - window: VecDeque::new(), + window: recovery.checkpoint.map(|c| c.records.into()).unwrap_or_default(), + sequence: recovery.sequence, }, grace: retention.map(|r| r.grace), + deletions, _timeline: broadcast, }) } @@ -215,6 +237,7 @@ impl Writer { mut commands, committer, grace, + mut deletions, _timeline, } = self; let shared = control.shared.clone(); @@ -228,7 +251,6 @@ impl Writer { let mut reads = FuturesUnordered::new(); let mut committer = Some(committer); let mut commit: Option> = None; - let mut deletions: VecDeque<(Instant, Vec)> = VecDeque::new(); let mut closed = false; // Cleared once the channel yields nothing more: every sender dropped, or it closed and drained. let mut accepting = true; @@ -263,11 +285,12 @@ impl Writer { None => accepting = false, Some(Command::Enroll { name, subscriber, recorder, timescale }) => { let (cancel, cancelled) = watch::channel(()); + let largest = shared.floors.get(&name).copied(); reads.push(guard(cancelled.clone(), recv(name.clone(), subscriber)).boxed()); tracks.insert(name, TrackState { recorder, timescale, - largest: None, + largest, reported: None, accepted: BTreeMap::new(), subscribed: true, @@ -406,7 +429,7 @@ impl Control { struct TrackState { recorder: Recorder, timescale: Timescale, - /// The newest group accepted; later arrivals must exceed it. + /// The newest group accepted or already stored; later arrivals must exceed it. largest: Option, /// The first-frame timestamp of the newest reported group; later groups must not precede it. reported: Option, @@ -652,6 +675,8 @@ struct Committer { retention: Option, /// Committed records still in the timeline window, oldest first. window: VecDeque, + /// Added to the timeline track's group sequences, continuing a resumed recording's numbering. + sequence: u64, } impl Committer { @@ -741,7 +766,8 @@ impl Committer { Poll::Pending => return Err(Error::Timeline("timeline group is still open".into())), } } - groups.push(convert(group.sequence, frames, timescale)?); + let sequence = group.sequence.checked_add(self.sequence).ok_or(Error::Overflow)?; + groups.push(convert(sequence, frames, timescale)?); } Ok(Object { groups }) } @@ -786,12 +812,14 @@ mod tests { }; use super::*; + use crate::store::list::Query; - /// An in-memory store whose group PUTs fail for one track. + /// An in-memory store whose group PUTs fail for one track, and whose listings fail at the end. #[derive(Debug, Clone)] struct Failing { inner: Arc, track: &'static str, + list: bool, } impl std::fmt::Display for Failing { @@ -837,7 +865,18 @@ mod tests { } fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result> { - self.inner.list(prefix) + let listed = self.inner.list(prefix); + match self.list { + true => listed + .chain(futures::stream::once(async { + Err(object_store::Error::NotImplemented { + operation: "list".into(), + implementer: "Failing".into(), + }) + })) + .boxed(), + false => listed, + } } async fn list_with_delimiter(&self, prefix: Option<&Path>) -> object_store::Result { @@ -995,6 +1034,7 @@ mod tests { let failing = Failing { inner: Arc::new(InMemory::new()), track: "audio", + list: false, }; let store = Store::new(failing, "rec"); let writer = Writer::new(store.clone(), source.consume(), Config::default()) @@ -1229,25 +1269,162 @@ mod tests { ); } - #[tokio::test] - async fn an_existing_recording_is_refused() { + /// Record `video` groups `sequences`, one per second, until the source ends. + async fn record(store: &Store, config: Config, sequences: std::ops::Range) { let source = broadcast::Info::new().produce(); let video = track(&source, "video"); - - let store = Store::new(InMemory::new(), "rec"); - let writer = Writer::new(store.clone(), source.consume(), Config::default()) - .await - .unwrap(); + let writer = Writer::new(store.clone(), source.consume(), config).await.unwrap(); writer.control().pacing_track("video").await.unwrap(); - group(&video, 0, &[0]); + for sequence in sequences { + group(&video, sequence, &[sequence * 1000, sequence * 1000 + 500]); + } video.finish().unwrap(); source.finish(); writer.run().await.unwrap(); + } + + /// Every stored group object, across all tracks. + async fn stored_groups(store: &Store) -> HashSet { + store + .list(&Query::new()) + .try_filter_map(|entry| async move { Ok(matches!(entry.key, Key::Groups { .. }).then_some(entry.key)) }) + .try_collect() + .await + .unwrap() + } + + /// The group objects the retained records advertise. + fn referenced(records: &[Record]) -> HashSet { + records + .iter() + .flat_map(|record| &record.tracks) + .map(|(name, ranges)| Key::groups(name.clone(), ranges[0].start..=ranges.last().unwrap().end).unwrap()) + .collect() + } + + fn orphan(sequence: u64) -> Object { + Object { + groups: vec![Group { + sequence, + frames: vec![Frame { + timestamp: sequence * 1000, + payload: "orphan".into(), + }], + }], + } + } + + #[tokio::test] + async fn a_restarted_writer_resumes_the_recording() { + let store = Store::new(InMemory::new(), "rec"); + record(&store, Config::default(), 0..3).await; + // The source's cache replays groups the recording already holds. + record(&store, Config::default(), 0..6).await; + + let records = window(&store).await; + assert_eq!( + records.iter().map(|record| record.segment).collect::>(), + (0..6).collect::>() + ); + assert_eq!(ranges(&records, "video"), (0..6).map(|s| (s, s)).collect::>()); + check_objects(&store, &records).await; - let again = broadcast::Info::new().produce(); - assert!(matches!( - Writer::new(store, again.consume(), Config::default()).await, - Err(Error::Occupied(_)) - )); + // The resumed timeline groups continue the stored numbering. + let mut sequences = Vec::new(); + for segment in 0..6 { + let object = store.get_segments(TIMELINE, segment).await.unwrap(); + sequences.extend(object.groups.iter().map(|group| group.sequence)); + } + assert!(sequences.windows(2).all(|pair| pair[0] < pair[1]), "{sequences:?}"); + } + + #[tokio::test] + async fn a_restarted_dvr_deletes_unreferenced_groups() { + let store = Store::new(InMemory::new(), "rec"); + let config = Config::default().with_retention(Retention::new(Duration::from_secs(2), Duration::ZERO)); + record(&store, config.clone(), 0..6).await; + + // An interrupted expiration, an uncommitted upload, and a track no retained record names. + store.put_groups("video", &orphan(1)).await.unwrap(); + store.put_groups("video", &orphan(7)).await.unwrap(); + store.put_groups("audio", &orphan(0)).await.unwrap(); + + // Groups at or below the uncommitted upload are refused, so nothing overlaps it. + record(&store, config, 6..10).await; + + let records = window(&store).await; + assert_eq!( + records.iter().map(|record| record.segment).collect::>(), + vec![5, 6, 7] + ); + assert_eq!(ranges(&records, "video"), vec![(5, 5), (8, 8), (9, 9)]); + check_objects(&store, &records).await; + assert_eq!(stored_groups(&store).await, referenced(&records)); + store.get_info("video").await.unwrap(); + store.get_info(TIMELINE).await.unwrap(); + store.get_segments(TIMELINE, 0).await.unwrap(); + } + + #[tokio::test] + async fn a_failed_recovery_deletes_nothing() { + let inner = Arc::new(InMemory::new()); + let store = Store::new(inner.clone(), "rec"); + let config = Config::default().with_retention(Retention::new(Duration::from_secs(2), Duration::ZERO)); + record(&store, config.clone(), 0..6).await; + store.put_groups("video", &orphan(1)).await.unwrap(); + let before = stored_groups(&store).await; + + let failing = Failing { + inner: inner.clone(), + track: "", + list: true, + }; + let source = broadcast::Info::new().produce(); + let result = Writer::new(Store::new(failing, "rec"), source.consume(), config.clone()).await; + assert!(matches!(result, Err(Error::Store(_)))); + + // A missing timeline object leaves the retained window unrecoverable. + store.delete(&Key::segments(TIMELINE, 3).unwrap()).await.unwrap(); + let result = Writer::new(store.clone(), source.consume(), config).await; + assert!(matches!(result, Err(Error::Timeline(_)))); + + assert_eq!(stored_groups(&store).await, before); + } + + #[tokio::test] + async fn a_dvr_window_longer_than_one_checkpoint_is_recovered() { + let store = Store::new(InMemory::new(), "rec"); + let config = Config::default().with_retention(Retention::new(Duration::from_secs(280), Duration::ZERO)); + record(&store, config, 0..300).await; + + let recovery = recover(&store, TIMELINE, true).await.unwrap(); + let checkpoint = recovery.checkpoint.unwrap(); + let records = window(&store).await; + assert!(records.len() > 256, "the window outgrows one checkpoint"); + assert_eq!(checkpoint.records, records); + assert_eq!(checkpoint.range.end, 300); + assert!(recovery.orphans.is_empty()); + assert_eq!(recovery.floors["video"], 299); + } + + #[tokio::test] + async fn a_timeline_that_does_not_end_at_the_next_segment_fails_recovery() { + let store = Store::new(InMemory::new(), "rec"); + record(&store, Config::default(), 0..6).await; + let before = stored_groups(&store).await; + + // `segments/5` still decodes, but it restates an earlier window, so resuming + // would write the next segment on top of it. + let older = store.get_segments(TIMELINE, 0).await.unwrap(); + store.delete(&Key::segments(TIMELINE, 5).unwrap()).await.unwrap(); + store.put_segments(TIMELINE, 5, &older).await.unwrap(); + + let source = broadcast::Info::new().produce(); + match Writer::new(store.clone(), source.consume(), Config::default()).await { + Err(Error::Timeline(message)) => assert!(message.contains("not segment 6"), "{message}"), + Err(err) => panic!("expected a timeline error, got {err}"), + Ok(_) => panic!("expected recovery to fail"), + } + assert_eq!(stored_groups(&store).await, before); } } diff --git a/rs/moq-json/src/window/encoder.rs b/rs/moq-json/src/window/encoder.rs index 6af1f43a5d..f06d52b988 100644 --- a/rs/moq-json/src/window/encoder.rs +++ b/rs/moq-json/src/window/encoder.rs @@ -86,6 +86,15 @@ impl ProducerConfig { } } +/// A retained window to continue, such as one replayed from stored groups. +#[derive(Debug, Clone, PartialEq)] +pub struct Checkpoint { + /// Absolute index of the oldest retained record, and of the next to be pushed. + pub range: std::ops::Range, + /// The newest retained records, oldest first, ending just before `range.end`. + pub records: Vec, +} + /// One encoded frame, and the group boundary it implies. #[derive(Clone, Debug)] #[non_exhaustive] @@ -407,6 +416,30 @@ impl Encoder { } impl Encoder { + /// Create an encoder continuing `checkpoint`, so the first edit opens a group restating it. + /// + /// Fails when the records outnumber the range, or the range exceeds the safe integer range. + pub fn resume(config: ProducerConfig, checkpoint: &Checkpoint) -> Result { + let Checkpoint { range, records } = checkpoint; + if range.start > range.end || range.end > MAX_INDEX || records.len() as u64 > range.end - range.start { + return Err(Error::Json("invalid window checkpoint".into())); + } + + let mut encoder = Self::new(config); + let skip = encoder + .config + .checkpoint_records + .map(|limit| records.len().saturating_sub(limit)) + .unwrap_or_default(); + encoder.window = records[skip..] + .iter() + .map(serde_json::to_value) + .collect::>()?; + encoder.offset = range.start; + encoder.start = range.end - encoder.window.len() as u64; + Ok(encoder) + } + /// Append one record to the back of the window. /// /// Emits a push into the open group, or a header restating the window (the new record included) diff --git a/rs/moq-json/src/window/mod.rs b/rs/moq-json/src/window/mod.rs index b318556194..51d22b18c2 100644 --- a/rs/moq-json/src/window/mod.rs +++ b/rs/moq-json/src/window/mod.rs @@ -55,7 +55,7 @@ mod producer; pub use consumer::Consumer; pub use decoder::{ConsumerConfig, Decoder, Event, Group}; -pub use encoder::{Encoded, Encoder, Pending, ProducerConfig}; +pub use encoder::{Checkpoint, Encoded, Encoder, Pending, ProducerConfig}; pub use producer::Producer; #[cfg(test)] @@ -291,6 +291,53 @@ mod test { ); } + #[test] + fn a_resumed_encoder_continues_the_window() { + let config = ProducerConfig::default().with_op_ratio(0).with_checkpoint_records(2); + let mut original = Encoder::::new(config.clone()); + let mut decoder = Decoder::::new(ConsumerConfig::default()); + for n in 0..4 { + let frame = original.push(&rec(n)).unwrap(); + decoder.group().decode(&frame.payload).unwrap(); + frame.commit(); + } + let frame = original.pop(1).unwrap().unwrap(); + decoder.group().decode(&frame.payload).unwrap(); + frame.commit(); + std::iter::from_fn(|| decoder.next_event()).for_each(drop); + + let checkpoint = Checkpoint { + range: 1..4, + records: vec![rec(1), rec(2), rec(3)], + }; + let mut resumed = Encoder::resume(config, &checkpoint).unwrap(); + assert_eq!(resumed.range(), 1..4); + assert_eq!( + resumed.window(), + vec![rec(2), rec(3)], + "trimmed to the checkpoint bound" + ); + + let frame = resumed.push(&rec(4)).unwrap(); + assert!(frame.keyframe, "the first edit restates the window"); + decoder.group().decode(&frame.payload).unwrap(); + frame.commit(); + assert_eq!( + std::iter::from_fn(|| decoder.next_event()).collect::>(), + vec![Event::Push { + index: 4, + value: rec(4) + }], + "a reader that kept up sees only the new record" + ); + + let invalid = Checkpoint { + range: 3..4, + records: vec![rec(2), rec(3)], + }; + assert!(Encoder::resume(ProducerConfig::default(), &invalid).is_err()); + } + #[test] fn pops_cross_the_omitted_checkpoint_prefix() { let config = ProducerConfig::default().with_checkpoint_records(2); diff --git a/rs/moq-json/src/window/producer.rs b/rs/moq-json/src/window/producer.rs index 5967f80bb5..50d604bf9c 100644 --- a/rs/moq-json/src/window/producer.rs +++ b/rs/moq-json/src/window/producer.rs @@ -6,7 +6,7 @@ use std::sync::{Arc, Mutex}; use serde::Serialize; use serde_json::Value; -use super::{Encoded, Encoder, ProducerConfig}; +use super::{Checkpoint, Encoded, Encoder, ProducerConfig}; use crate::Result; /// Publishes a sliding window of JSON records over a track. @@ -32,13 +32,17 @@ impl Clone for Producer { impl Producer { /// Create a producer that publishes to the given track. pub fn new(track: moq_net::track::Producer, config: ProducerConfig) -> Self { + Self::with_encoder(track, Encoder::new(config)) + } + + fn with_encoder(track: moq_net::track::Producer, encoder: Encoder) -> Self { Self { inner: Arc::new(Mutex::new(Inner { track: Track { inner: track, group: None, }, - encoder: Encoder::new(config), + encoder, finished: false, })), _marker: PhantomData, @@ -89,6 +93,14 @@ impl Producer { } impl Producer { + /// Create a producer continuing `checkpoint` on the given track. + /// + /// The first edit opens a group restating the checkpoint, so a reader that already holds those + /// records sees only what follows. Fails like [`Encoder::resume`]. + pub fn resume(track: moq_net::track::Producer, config: ProducerConfig, checkpoint: &Checkpoint) -> Result { + Ok(Self::with_encoder(track, Encoder::resume(config, checkpoint)?)) + } + /// Append one record to the back of the window. pub fn push(&mut self, value: &T) -> Result<()> { self.inner.lock().unwrap().push(value) diff --git a/rs/moq-mux/src/error.rs b/rs/moq-mux/src/error.rs index 5a1637f7a0..f107b735c6 100644 --- a/rs/moq-mux/src/error.rs +++ b/rs/moq-mux/src/error.rs @@ -200,6 +200,11 @@ pub enum Error { #[error("timeline segment {0} was not yielded for deferred publication")] TimelineDeferredRecord(u64), + /// [`timeline::Producer::resume`](crate::timeline::Producer::resume) received a checkpoint + /// whose record at this window index is a different segment. + #[error("timeline checkpoint record at index {0} is a different segment")] + TimelineCheckpoint(u64), + /// Error from a muxer/demuxer that reports via `anyhow` (currently MPEG-TS). /// Boxed in an `Arc` so the enum stays `Clone` (`anyhow::Error` is not). #[error("{0}")] diff --git a/rs/moq-mux/src/timeline.rs b/rs/moq-mux/src/timeline.rs index ecc240c092..669df15f49 100644 --- a/rs/moq-mux/src/timeline.rs +++ b/rs/moq-mux/src/timeline.rs @@ -57,6 +57,7 @@ use std::time::Duration; use hang::catalog::Archive; use hang::timeline::{DEFAULT_NAME, Range, Record, RecordExt}; +use moq_json::window::Checkpoint; use moq_net::{Timescale, Timestamp}; @@ -739,16 +740,30 @@ struct Output { impl Output { fn prepare(&mut self) -> crate::Result<()> { if self.sink.is_none() && !self.closed { - let info = moq_net::track::Info::default().with_priority(hang::catalog::PRIORITY.catalog); - let net = self.broadcast.create_track(DEFAULT_NAME, info)?; - let config = moq_json::window::ProducerConfig::default() - .with_compression(true) - .with_checkpoint_records(CHECKPOINT_RECORDS); - self.sink = Some(moq_json::window::Producer::new(net, config)); + let net = self.create_track()?; + self.sink = Some(moq_json::window::Producer::new(net, Self::config())); } Ok(()) } + /// Create the timeline track now, continuing `checkpoint`. + fn resume(&mut self, checkpoint: &Checkpoint) -> crate::Result<()> { + let net = self.create_track()?; + self.sink = Some(moq_json::window::Producer::resume(net, Self::config(), checkpoint)?); + Ok(()) + } + + fn create_track(&self) -> crate::Result { + let info = moq_net::track::Info::default().with_priority(hang::catalog::PRIORITY.catalog); + Ok(self.broadcast.create_track(DEFAULT_NAME, info)?) + } + + fn config() -> moq_json::window::ProducerConfig { + moq_json::window::ProducerConfig::default() + .with_compression(true) + .with_checkpoint_records(CHECKPOINT_RECORDS) + } + fn push(&mut self, record: &Record) -> crate::Result<()> { self.prepare()?; let Some(sink) = self.sink.as_mut() else { @@ -824,6 +839,31 @@ impl Producer { } } + /// A timeline for `broadcast` continuing `checkpoint`, such as a window recovered from storage. + /// + /// A timeline's window index is its segment number, so the next record is segment + /// `checkpoint.range.end`. Creates the timeline track immediately; its first group restates the + /// checkpoint. Fails when a checkpoint record is not the segment at its index, the checkpoint + /// is malformed, or the track cannot be created. + pub fn resume( + broadcast: &moq_net::broadcast::Producer, + config: Config, + checkpoint: &Checkpoint, + ) -> crate::Result { + let producer = Self::new(broadcast, config); + producer.output.lock().unwrap().resume(checkpoint)?; + + // The encoder accepted the checkpoint, so the records fit before `range.end`. + let start = checkpoint.range.end - checkpoint.records.len() as u64; + for (index, record) in (start..).zip(&checkpoint.records) { + if record.segment != index { + return Err(crate::Error::TimelineCheckpoint(index)); + } + } + producer.segmenter.state.lock().unwrap().next_segment = checkpoint.range.end; + Ok(producer) + } + /// Enroll `name` without letting it influence segmentation, returning its [`Recorder`]. /// /// Its groups are recorded into whichever segment is open when they arrive, but the track @@ -1322,6 +1362,46 @@ mod test { (broadcast, timeline) } + #[tokio::test] + async fn a_resumed_timeline_continues_the_checkpoint() { + let broadcast = moq_net::broadcast::Info::new().produce(); + let checkpoint = Checkpoint { + range: 1..3, + records: vec![Record::new(1, 1_000, 1_000), Record::new(2, 2_000, 1_000)], + }; + let mut timeline = Producer::resume(&broadcast, Config::default(), &checkpoint).unwrap(); + + let mut video = timeline.pacing_track("video0").unwrap(); + video.record(10, ms(3_000), true); + video.record(11, ms(4_000), true); + video.end(ms(5_000)); + drop(video); + timeline.finish().unwrap(); + + assert_eq!( + drain(&broadcast, &timeline).await, + vec![ + entry(1, 1_000, 1_000, &[]), + entry(2, 2_000, 1_000, &[]), + entry(3, 3_000, 1_000, &[("video0", &[(10, 10)])]), + entry(4, 4_000, 1_000, &[("video0", &[(11, 11)])]), + ] + ); + } + + #[test] + fn a_checkpoint_record_must_be_the_segment_at_its_index() { + let broadcast = moq_net::broadcast::Info::new().produce(); + let checkpoint = Checkpoint { + range: 0..1, + records: vec![Record::new(5, 0, 1_000)], + }; + assert!(matches!( + Producer::resume(&broadcast, Config::default(), &checkpoint), + Err(crate::Error::TimelineCheckpoint(0)) + )); + } + #[tokio::test] async fn deferred_records_are_invisible_until_committed() { let (broadcast, timeline) = setup();