From 689315e33553a09bc7a1f0cb59af82ad6c93dee3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 09:41:14 -0700 Subject: [PATCH 1/3] quest(archive): claim DVR rewind Co-Authored-By: Claude Opus 5.5 From 42dc7631bb6450d23a4c90b8ec73009af167ed8d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:06:15 -0700 Subject: [PATCH 2/3] feat(archive): rewind a track into its recording, then splice to live Co-Authored-By: Claude Opus 5.5 --- doc/lib/rs/index.md | 2 +- quest/m1/archive/README.md | 8 +- quest/m1/archive/dvr.md | 31 --- rs/moq-archive/README.md | 4 + rs/moq-archive/src/{reader => }/index.rs | 84 +++++++- rs/moq-archive/src/lib.rs | 7 +- rs/moq-archive/src/reader/mod.rs | 4 +- rs/moq-archive/src/rewind/mod.rs | 236 +++++++++++++++++++++++ rs/moq-archive/src/rewind/tests.rs | 214 ++++++++++++++++++++ rs/moq-archive/src/writer.rs | 41 +++- 10 files changed, 586 insertions(+), 45 deletions(-) delete mode 100644 quest/m1/archive/dvr.md rename rs/moq-archive/src/{reader => }/index.rs (68%) create mode 100644 rs/moq-archive/src/rewind/mod.rs create mode 100644 rs/moq-archive/src/rewind/tests.rs diff --git a/doc/lib/rs/index.md b/doc/lib/rs/index.md index 2a02438809..979bb2fbbc 100644 --- a/doc/lib/rs/index.md +++ b/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. | diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index fabafaf0e1..853ed01b4b 100644 --- a/quest/m1/archive/README.md +++ b/quest/m1/archive/README.md @@ -54,6 +54,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 @@ -122,7 +129,6 @@ owned by that prerequisite, not duplicated in archive storage. - [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 ## Related diff --git a/quest/m1/archive/dvr.md b/quest/m1/archive/dvr.md deleted file mode 100644 index 096633a809..0000000000 --- a/quest/m1/archive/dvr.md +++ /dev/null @@ -1,31 +0,0 @@ -# [M] DVR rewind - -## Goal - -A viewer seeks through a bounded archive and returns to live playback using the -same timeline and group-range objects as an unbounded archive. - -## Plan - -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 -second media format. Missing groups remain ordinary gaps. - -An unbounded archive can continue the same segment numbering without rewriting -objects retained from an earlier DVR window. - -Test seeks within the retained window, expiry during a seek, missing groups, -restart recovery, and return to live without duplicated or rewound playback. -Use the writer/reader fixtures; a retention defect is fixed in its owning layer. - -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`. - -## Closes - -- [#2275](https://github.com/moq-dev/moq/issues/2275) - close this issue when the quest finishes diff --git a/rs/moq-archive/README.md b/rs/moq-archive/README.md index a357d1f807..283579294f 100644 --- a/rs/moq-archive/README.md +++ b/rs/moq-archive/README.md @@ -14,6 +14,10 @@ that needs runtime dispatch passes `Arc`. 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 ``` diff --git a/rs/moq-archive/src/reader/index.rs b/rs/moq-archive/src/index.rs similarity index 68% rename from rs/moq-archive/src/reader/index.rs rename to rs/moq-archive/src/index.rs index bedbd7b9e2..d0d1e5c0de 100644 --- a/rs/moq-archive/src/reader/index.rs +++ b/rs/moq-archive/src/index.rs @@ -1,5 +1,5 @@ use std::collections::{BTreeMap, HashMap}; -use std::ops::{Range, RangeInclusive}; +use std::ops::{Bound, Range, RangeInclusive}; use hang::timeline::Record; @@ -7,16 +7,18 @@ 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, /// The advertised runs, ascending and nonoverlapping; groups between them never existed. pub runs: Vec>, + /// 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 { + fn new(ranges: &[hang::timeline::Range], pts: u64) -> Option { let mut runs: Vec> = 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>, /// 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 { + 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 { + 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 { 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(); diff --git a/rs/moq-archive/src/lib.rs b/rs/moq-archive/src/lib.rs index 3d5d9bb1d8..f97639f95e 100644 --- a/rs/moq-archive/src/lib.rs +++ b/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`; 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; diff --git a/rs/moq-archive/src/reader/mod.rs b/rs/moq-archive/src/reader/mod.rs index 64b22c9fab..437165f9bd 100644 --- a/rs/moq-archive/src/reader/mod.rs +++ b/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}; diff --git a/rs/moq-archive/src/rewind/mod.rs b/rs/moq-archive/src/rewind/mod.rs new file mode 100644 index 0000000000..b47fafc181 --- /dev/null +++ b/rs/moq-archive/src/rewind/mod.rs @@ -0,0 +1,236 @@ +//! Seek a track back into its recording, then follow it live again. +//! +//! A [`Rewind`] pairs a live broadcast with the broadcast replaying its recording, which a +//! [`Reader`](crate::Reader) serves (the catalog `archive` entry's `replay`). [`Rewind::seek`] +//! reads the recording's timeline, starts at the segment containing a timestamp, and FETCHes every +//! group the timeline advertises in order. Once the next group is newer than anything advertised, +//! the track subscribes to the live broadcast from that group on. The recording keeps the source's +//! group sequences, so the splice neither repeats nor rewinds a group. +//! +//! The timeline decides what is requested. A group it never advertised, one that expired before +//! its turn, or one whose FETCH is not found is skipped like any other gap. +//! +//! ```no_run +//! # async fn example(live: moq_net::broadcast::Consumer, replay: moq_net::broadcast::Consumer) -> moq_archive::Result<()> { +//! use moq_archive::Rewind; +//! +//! let timeline = hang::catalog::Archive::new(hang::timeline::DEFAULT_NAME); +//! let rewind = Rewind::new(live, replay, timeline); +//! let mut video = rewind.seek("video", moq_net::Timestamp::from_secs(30)?).await?; +//! while let Some(group) = video.next_group().await? { +//! // Decode `group`; `video.is_live()` turns true once playback reaches the live broadcast. +//! # drop(group); +//! } +//! # Ok(()) +//! # } +//! ``` + +use std::task::{Poll, ready}; + +use hang::catalog; +use hang::timeline::Record; +use moq_json::window; +use moq_net::{Timescale, Timestamp, broadcast, group, track}; + +use crate::index::Index; +use crate::{Error, Result}; + +/// Pairs a live broadcast with the broadcast replaying its recording. +#[derive(Clone)] +pub struct Rewind { + live: broadcast::Consumer, + archive: broadcast::Consumer, + timeline: catalog::Archive, +} + +impl Rewind { + /// Rewind `live` through `archive`, the broadcast replaying its recording, whose timeline + /// `timeline` names. + pub fn new(live: broadcast::Consumer, archive: broadcast::Consumer, timeline: catalog::Archive) -> Self { + Self { + live, + archive, + timeline, + } + } + + /// Read `track` from the recorded segment containing `at`, then from the live broadcast. + /// + /// Starts at the first group of the newest retained segment starting at or before `at`, or + /// of the oldest one when `at` predates the window. Waits for the timeline's first record; a + /// track the timeline never names, or a timeline that ends empty, starts at the live edge. + pub async fn seek(&self, track: &str, at: Timestamp) -> Result { + let scale = self.timeline.timescale as u64; + let timescale = Timescale::new(scale).map_err(|_| Error::Timescale(scale))?; + let at = at.convert(timescale).map_err(|_| Error::Overflow)?.value(); + + let subscriber = self.archive.track(&self.timeline.track)?.subscribe(None).await?; + let config = window::ConsumerConfig::default().with_compression(true); + let mut timeline = window::Consumer::new(subscriber, config); + let first = timeline.next().await; + + let mut archive = Archive { + name: track.to_string(), + track: self.archive.track(track)?, + timeline: Some(timeline), + index: Index::default(), + next: 0, + fetching: None, + }; + // The first event opens the window's checkpoint; the rest of it is already here. + archive.apply(first)?; + archive.poll_timeline(&kio::Waiter::noop())?; + + let live = self.live.track(track)?; + let source = match archive.index.seek(track, at) { + Some(next) => { + archive.next = next; + Source::Archive(archive) + } + None => Source::Subscribing(subscribe(&live, None)), + }; + Ok(Track { live, source }) + } +} + +/// One track read from the recording, then from the live broadcast. +pub struct Track { + live: track::Consumer, + source: Source, +} + +enum Source { + Archive(Archive), + Subscribing(Subscribing), + Live(track::Ordered), +} + +struct Subscribing { + pending: track::Subscribing, + /// The first group the live broadcast may deliver, or `None` for its live edge. + floor: Option, +} + +impl Track { + /// Whether groups now come from the live broadcast rather than the recording. + pub fn is_live(&self) -> bool { + !matches!(self.source, Source::Archive(_)) + } + + /// Poll for the next group, each with a higher sequence than the last, without blocking. + /// + /// Returns `None` once the live track finishes. Fails when the recording's timeline or a + /// track errors; a group that is merely unavailable is skipped instead. + pub fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + loop { + match &mut self.source { + Source::Archive(archive) => { + if let Some(group) = ready!(archive.poll_next_group(waiter))? { + return Poll::Ready(Ok(Some(group))); + } + self.source = Source::Subscribing(subscribe(&self.live, Some(archive.next))); + } + Source::Subscribing(subscribing) => { + let mut live = ready!(subscribing.pending.poll_ok(waiter))?.ordered(); + if let Some(floor) = subscribing.floor { + live.set_groups(floor..); + } + self.source = Source::Live(live); + } + Source::Live(live) => return Poll::Ready(Ok(ready!(live.poll_next_group(waiter))?)), + } + } + } + + /// Return the next group, each with a higher sequence than the last. + pub async fn next_group(&mut self) -> Result> { + kio::wait(|waiter| self.poll_next_group(waiter)).await + } +} + +/// Subscribe to the live track from `floor`, or from its live edge when `None`. +fn subscribe(live: &track::Consumer, floor: Option) -> Subscribing { + let subscription = match floor { + // Reach back as far as the publisher keeps groups, but never below the floor. + Some(floor) => track::Subscription::default() + .with_max_age(crate::writer::REPLAY) + .with_start(track::Position::group(floor)), + None => track::Subscription::default(), + }; + Subscribing { + pending: live.subscribe(subscription).into_inner(), + floor, + } +} + +/// The recording side of a [`Track`]: the timeline it follows and the next group to FETCH. +struct Archive { + name: String, + /// The track on the replay broadcast. + track: track::Consumer, + /// The recording's timeline, until it ends. + timeline: Option>, + index: Index, + /// The lowest group not yet returned or skipped. + next: u64, + fetching: Option<(u64, track::Fetching)>, +} + +impl Archive { + /// The next advertised group, or `None` once `next` is past everything advertised. + fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + loop { + self.poll_timeline(waiter)?; + + let Some((sequence, fetching)) = &self.fetching else { + let Some(sequence) = self.index.next(&self.name, self.next) else { + return Poll::Ready(Ok(None)); + }; + let fetching = self.track.fetch_group(sequence, None).into_inner(); + self.fetching = Some((sequence, fetching)); + continue; + }; + + let sequence = *sequence; + let result = ready!(kio::Pollable::poll(fetching, waiter)); + self.fetching = None; + // Advertised sequences stay within the recording's id range, so this cannot overflow. + self.next = sequence + 1; + match result { + Ok(group) => return Poll::Ready(Ok(Some(group))), + // Expired or unreadable since the timeline advertised it: an ordinary gap. + Err(moq_net::Error::NotFound) => { + tracing::debug!(track = self.name, sequence, "skipping an unavailable archived group"); + } + Err(err) => return Poll::Ready(Err(err.into())), + } + } + } + + /// Apply every timeline event that is ready. + fn poll_timeline(&mut self, waiter: &kio::Waiter) -> Result<()> { + while let Some(timeline) = &mut self.timeline { + let Poll::Ready(event) = timeline.poll_next(waiter) else { + break; + }; + self.apply(event)?; + } + Ok(()) + } + + fn apply(&mut self, event: moq_json::Result>>) -> Result<()> { + match event.map_err(|err| Error::Timeline(err.to_string()))? { + Some(window::Event::Push { index, value }) => self.index.push(index, &value), + Some(window::Event::Pop(range)) => { + self.index.pop(range); + } + Some(_) => {} + // Following stops; the advertised groups stay servable. + None => self.timeline = None, + } + Ok(()) + } +} + +#[cfg(test)] +mod tests; diff --git a/rs/moq-archive/src/rewind/tests.rs b/rs/moq-archive/src/rewind/tests.rs new file mode 100644 index 0000000000..8aa5c24f29 --- /dev/null +++ b/rs/moq-archive/src/rewind/tests.rs @@ -0,0 +1,214 @@ +use std::time::Duration; + +use bytes::Bytes; +use object_store::memory::InMemory; + +use super::*; +use crate::writer::{Config, Retention}; +use crate::{Key, Reader, Store, Writer, reader}; + +const TIMELINE: &str = hang::timeline::DEFAULT_NAME; + +fn ms(v: u64) -> Timestamp { + Timestamp::from_millis(v).unwrap() +} + +/// A live source recorded by a [`Writer`] and replayed by a [`Reader`], as a DVR deployment runs. +struct Dvr { + store: Store, + source: broadcast::Producer, + video: track::Producer, + replay: broadcast::Producer, + reader: Reader, +} + +impl Dvr { + async fn new(config: Config) -> Self { + let store = Store::new(InMemory::new(), "rec"); + let (source, video) = record(&store, config).await; + let replay = broadcast::Info::new().produce(); + let reader = Reader::open(store.clone(), &replay, reader::Config::new(TIMELINE)) + .await + .unwrap(); + tokio::spawn(reader.serve()); + Self { + store, + source, + video, + replay, + reader, + } + } + + /// Write one second-long group per sequence. + fn write(&self, sequences: impl IntoIterator) { + for sequence in sequences { + let mut group = self.video.create_group(group::Info { sequence }).unwrap(); + for timestamp in [sequence * 1000, sequence * 1000 + 500] { + group.write_frame(ms(timestamp), payload(sequence, timestamp)).unwrap(); + } + group.finish().unwrap(); + } + } + + /// Wait for the writer to store timeline `segment`, then replay it. + async fn commit(&mut self, segment: u64) { + while self.store.get_segments(TIMELINE, segment).await.is_err() { + tokio::task::yield_now().await; + } + self.reader.refresh().await.unwrap(); + } + + fn rewind(&self) -> Rewind { + Rewind::new( + self.source.consume(), + self.replay.consume(), + catalog::Archive::new(TIMELINE), + ) + } +} + +/// Start recording a fresh source's `video` track into `store`, resuming what it already holds. +async fn record(store: &Store, config: Config) -> (broadcast::Producer, track::Producer) { + let source = broadcast::Info::new().produce(); + let info = track::Info::default() + .with_timescale(Timescale::MILLI) + .with_max_age(Duration::from_secs(3600)); + let video = source.create_track("video", info).unwrap(); + let writer = Writer::new(store.clone(), source.consume(), config).await.unwrap(); + writer.control().pacing_track("video").await.unwrap(); + tokio::spawn(writer.run()); + (source, video) +} + +fn payload(sequence: u64, timestamp: u64) -> Bytes { + Bytes::from(format!("{sequence}@{timestamp}")) +} + +fn dvr() -> Config { + Config::default().with_retention(Retention::new(Duration::from_secs(4), Duration::ZERO)) +} + +/// Read the next group, checking its frames are the source's, and report where it came from. +async fn next(track: &mut Track) -> (u64, bool) { + let mut group = track.next_group().await.unwrap().expect("the track is still live"); + let sequence = group.sequence; + let mut frames = Vec::new(); + while let Some(frame) = group.read_frame().await.unwrap() { + frames.push((frame.timestamp, frame.payload)); + } + let expected: Vec<_> = [sequence * 1000, sequence * 1000 + 500] + .into_iter() + .map(|timestamp| (ms(timestamp), payload(sequence, timestamp))) + .collect(); + assert_eq!(frames, expected, "group {sequence}"); + (sequence, track.is_live()) +} + +#[tokio::test] +async fn a_seek_plays_the_recording_then_splices_to_live() { + let mut dvr = Dvr::new(dvr()).await; + dvr.write(0..10); + // Group 9 is still open in its segment, so the recording ends at group 8. + dvr.commit(8).await; + + let mut video = dvr.rewind().seek("video", ms(6_200)).await.unwrap(); + assert!(!video.is_live()); + assert_eq!(next(&mut video).await, (6, false)); + assert_eq!(next(&mut video).await, (7, false)); + assert_eq!(next(&mut video).await, (8, false)); + + // Live runs ahead of what the recording has replayed, so the splice reaches back into the + // live cache instead of jumping to its newest group. + dvr.write(10..12); + for sequence in 9..12 { + assert_eq!(next(&mut video).await, (sequence, true)); + } +} + +#[tokio::test] +async fn a_seek_before_the_window_starts_at_its_oldest_segment() { + let mut dvr = Dvr::new(dvr()).await; + dvr.write(0..10); + dvr.commit(8).await; + + // Four seconds of one-second segments: 5 through 8 are retained. + let mut video = dvr.rewind().seek("video", ms(0)).await.unwrap(); + assert_eq!(next(&mut video).await, (5, false)); +} + +#[tokio::test] +async fn expiry_during_a_seek_skips_to_the_retained_window() { + let mut dvr = Dvr::new(dvr()).await; + dvr.write(0..10); + dvr.commit(8).await; + + let mut video = dvr.rewind().seek("video", ms(0)).await.unwrap(); + assert_eq!(next(&mut video).await, (5, false)); + + // The viewer pauses while the window moves past it, popping 5 through 8. + dvr.write(10..14); + dvr.commit(12).await; + + // The timeline, not the replay track's cache of group 5, decides what comes next. + assert_eq!(next(&mut video).await, (9, false)); + assert_eq!(next(&mut video).await, (10, false)); +} + +#[tokio::test] +async fn missing_groups_are_gaps() { + let mut dvr = Dvr::new(Config::default()).await; + // The source never produced group 3. + dvr.write([0, 1, 2, 4, 5, 6]); + dvr.commit(4).await; + // Group 1's object is lost after its record was committed. + dvr.store.delete(&Key::groups("video", 1..=1).unwrap()).await.unwrap(); + + let mut video = dvr.rewind().seek("video", ms(0)).await.unwrap(); + assert_eq!(next(&mut video).await, (0, false)); + assert_eq!(next(&mut video).await, (2, false)); + assert_eq!(next(&mut video).await, (4, false)); + assert_eq!(next(&mut video).await, (5, false)); + assert_eq!(next(&mut video).await, (6, true)); +} + +#[tokio::test] +async fn a_seek_spans_a_writer_restart() { + let mut dvr = Dvr::new(Config::default()).await; + dvr.write(0..4); + dvr.commit(2).await; + dvr.video.finish().unwrap(); + dvr.source.finish(); + // The clean end flushes group 3 as the final segment. + dvr.commit(3).await; + + // The publisher restarts, continuing its sequences, and a new writer resumes the recording. + let (source, video) = record(&dvr.store, Config::default()).await; + dvr.source = source; + dvr.video = video; + dvr.write(4..9); + dvr.commit(7).await; + + let mut video = dvr.rewind().seek("video", ms(1_000)).await.unwrap(); + for sequence in 1..8 { + assert_eq!(next(&mut video).await, (sequence, false)); + } + assert_eq!(next(&mut video).await, (8, true)); +} + +#[tokio::test] +async fn a_track_the_recording_never_names_starts_live() { + let mut dvr = Dvr::new(Config::default()).await; + dvr.write(0..3); + dvr.commit(1).await; + + let info = track::Info::default().with_timescale(Timescale::MILLI); + let chat = dvr.source.create_track("chat", info).unwrap(); + let mut track = dvr.rewind().seek("chat", ms(0)).await.unwrap(); + assert!(track.is_live()); + + let mut group = chat.append_group().unwrap(); + group.write_frame(ms(0), "hi").unwrap(); + group.finish().unwrap(); + assert_eq!(track.next_group().await.unwrap().unwrap().sequence, 0); +} diff --git a/rs/moq-archive/src/writer.rs b/rs/moq-archive/src/writer.rs index 5f05cdc083..c85d14211f 100644 --- a/rs/moq-archive/src/writer.rs +++ b/rs/moq-archive/src/writer.rs @@ -44,7 +44,7 @@ use crate::segment::{Frame, Group, Object}; use crate::{Error, Info, Key, Result, Store}; /// Subscribers ask for every cached group; the publisher clamps this to its own max age. -const REPLAY: Duration = Duration::from_secs(u32::MAX as u64); +pub(crate) const REPLAY: Duration = Duration::from_secs(u32::MAX as u64); /// How a [`Writer`] segments and retains its recording. #[derive(Clone, Debug, Default)] @@ -804,6 +804,7 @@ mod tests { use futures::TryStreamExt; use futures::stream::BoxStream; + use object_store::ObjectStoreExt; use object_store::memory::InMemory; use object_store::path::Path; use object_store::{ @@ -1365,6 +1366,44 @@ mod tests { store.get_segments(TIMELINE, 0).await.unwrap(); } + #[tokio::test] + async fn an_archive_continues_a_dvr_without_rewriting_it() { + let store = Store::new(InMemory::new(), "rec"); + let dvr = Config::default().with_retention(Retention::new(Duration::from_secs(2), Duration::ZERO)); + record(&store, dvr, 0..6).await; + let retained = window(&store).await; + let mut objects = BTreeMap::new(); + for key in referenced(&retained) { + let path = store.path(&key).unwrap(); + objects.insert( + path.clone(), + store.inner().get(&path).await.unwrap().bytes().await.unwrap(), + ); + } + for segment in 0..6 { + let path = store.path(&Key::segments(TIMELINE, segment).unwrap()).unwrap(); + objects.insert( + path.clone(), + store.inner().get(&path).await.unwrap().bytes().await.unwrap(), + ); + } + + // Without retention the restart keeps the DVR's window and never pops again. + record(&store, Config::default(), 6..10).await; + + let records = window(&store).await; + assert_eq!( + records.iter().map(|record| record.segment).collect::>(), + (3..10).collect::>() + ); + assert_eq!(ranges(&records, "video"), (3..10).map(|s| (s, s)).collect::>()); + check_objects(&store, &records).await; + for (path, bytes) in objects { + let stored = store.inner().get(&path).await.unwrap().bytes().await.unwrap(); + assert_eq!(stored, bytes, "{path} was rewritten"); + } + } + #[tokio::test] async fn a_failed_recovery_deletes_nothing() { let inner = Arc::new(InMemory::new()); From 8fc4825468050ddf7ee50cc8967ba13bd54e77c2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:31:35 -0700 Subject: [PATCH 3/3] fix(archive): keep the rewind source unboxed Co-Authored-By: Claude Opus 5.5 --- rs/moq-archive/src/rewind/mod.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rs/moq-archive/src/rewind/mod.rs b/rs/moq-archive/src/rewind/mod.rs index b47fafc181..7d357fc47b 100644 --- a/rs/moq-archive/src/rewind/mod.rs +++ b/rs/moq-archive/src/rewind/mod.rs @@ -99,6 +99,8 @@ pub struct Track { source: Source, } +// One per track, moved between states in place, so the variants' sizes don't matter. +#[allow(clippy::large_enum_variant)] enum Source { Archive(Archive), Subscribing(Subscribing),