From fd25766d01e8244d4a6ce24676df5bfec5dc106f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:04:39 -0700 Subject: [PATCH 01/10] quest(archive): claim proof Co-Authored-By: Claude Opus 5.5 From 6cd9a00e0db62aa79f45577307dd35df66795c55 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:17:10 -0700 Subject: [PATCH 02/10] test(archive): share one instrumented test object store Co-Authored-By: Claude Opus 5.5 --- rs/moq-archive/src/lib.rs | 2 + rs/moq-archive/src/mock.rs | 256 +++++++++++++++++++++++++++++ rs/moq-archive/src/reader/tests.rs | 90 ++-------- rs/moq-archive/src/store.rs | 104 +----------- rs/moq-archive/src/writer.rs | 101 +----------- 5 files changed, 279 insertions(+), 274 deletions(-) create mode 100644 rs/moq-archive/src/mock.rs diff --git a/rs/moq-archive/src/lib.rs b/rs/moq-archive/src/lib.rs index 3d5d9bb1d8..f84d4c8a01 100644 --- a/rs/moq-archive/src/lib.rs +++ b/rs/moq-archive/src/lib.rs @@ -17,6 +17,8 @@ pub use object_store; mod error; pub mod info; +#[cfg(test)] +mod mock; mod path; pub mod reader; mod recover; diff --git a/rs/moq-archive/src/mock.rs b/rs/moq-archive/src/mock.rs new file mode 100644 index 0000000000..1869b1455c --- /dev/null +++ b/rs/moq-archive/src/mock.rs @@ -0,0 +1,256 @@ +//! A test object store: any inner store plus an operation log, injected failures, and +//! S3-style paginated listing. + +use std::sync::{Arc, Mutex}; + +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +use object_store::list::{PaginatedListOptions, PaginatedListResult, PaginatedListStore}; +use object_store::memory::InMemory; +use object_store::path::Path; +use object_store::{ + CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, PutMultipartOptions, + PutOptions, PutPayload, PutResult, +}; + +/// S3's ListObjectsV2 page limit. +const MAX_KEYS: usize = 1000; + +/// One call made against a [`Mock`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum Op { + Get(String), + Put(String), + Delete(String), + /// A streaming or paginated listing, with its prefix and exclusive offset. + List { prefix: String, offset: Option }, +} + +#[derive(Debug, Default)] +struct State { + ops: Vec, + /// PUTs whose path contains any of these fail. + fail_puts: Vec, + /// GETs whose path contains any of these return Not Found. + hide_gets: Vec, + /// Streaming listings end with an error after every entry. + fail_lists: bool, + /// Streaming listings yield entries in descending order, like a backend that promises none. + unordered: bool, +} + +#[derive(Debug, Clone)] +pub(crate) struct Mock { + inner: Arc, + state: Arc>, +} + +impl Mock { + pub fn new(inner: impl ObjectStore) -> Self { + Self { + inner: Arc::new(inner), + state: Default::default(), + } + } + + pub fn memory() -> Self { + Self::new(InMemory::new()) + } + + /// Share this store's objects, without its log or failures. + pub fn fork(&self) -> Self { + Self { + inner: self.inner.clone(), + state: Default::default(), + } + } + + /// Every call since the last take. + pub fn take(&self) -> Vec { + std::mem::take(&mut self.state().ops) + } + + /// Paths of every GET since the last take. + pub fn gets(&self) -> Vec { + self.take() + .into_iter() + .filter_map(|op| match op { + Op::Get(path) => Some(path), + _ => None, + }) + .collect() + } + + pub fn fail_puts(&self, pattern: &str) -> &Self { + self.state().fail_puts.push(pattern.to_string()); + self + } + + pub fn hide_gets(&self, pattern: &str) -> &Self { + self.state().hide_gets.push(pattern.to_string()); + self + } + + pub fn fail_lists(&self) -> &Self { + self.state().fail_lists = true; + self + } + + pub fn unordered(&self) -> &Self { + self.state().unordered = true; + self + } + + /// Clear every injected failure. + pub fn heal(&self) { + let mut state = self.state(); + state.fail_puts.clear(); + state.hide_gets.clear(); + state.fail_lists = false; + } + + fn state(&self) -> std::sync::MutexGuard<'_, State> { + self.state.lock().unwrap() + } + + fn log(&self, op: Op) { + self.state().ops.push(op); + } + + fn listed(&self, prefix: Option<&Path>, offset: Option<&Path>) -> BoxStream<'static, object_store::Result> { + self.log(Op::List { + prefix: prefix.map(ToString::to_string).unwrap_or_default(), + offset: offset.map(ToString::to_string), + }); + let listed = match offset { + Some(offset) => self.inner.list_with_offset(prefix, offset), + None => self.inner.list(prefix), + }; + let state = self.state(); + let listed = match state.unordered { + true => futures::stream::once(async move { + let mut metas: Vec = listed.try_collect().await?; + metas.sort_by(|a, b| b.location.cmp(&a.location)); + Ok::<_, object_store::Error>(futures::stream::iter(metas.into_iter().map(Ok))) + }) + .try_flatten() + .boxed(), + false => listed, + }; + match state.fail_lists { + true => listed + .chain(futures::stream::once(async { Err(unsupported("list")) })) + .boxed(), + false => listed, + } + } +} + +fn unsupported(operation: &str) -> object_store::Error { + object_store::Error::NotImplemented { + operation: operation.into(), + implementer: "Mock".into(), + } +} + +impl std::fmt::Display for Mock { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Mock({})", self.inner) + } +} + +#[async_trait::async_trait] +impl ObjectStore for Mock { + async fn put_opts(&self, location: &Path, payload: PutPayload, opts: PutOptions) -> object_store::Result { + self.log(Op::Put(location.to_string())); + if self.state().fail_puts.iter().any(|p| location.as_ref().contains(p)) { + return Err(unsupported("put")); + } + self.inner.put_opts(location, payload, opts).await + } + + async fn put_multipart_opts( + &self, + location: &Path, + opts: PutMultipartOptions, + ) -> object_store::Result> { + self.inner.put_multipart_opts(location, opts).await + } + + async fn get_opts(&self, location: &Path, options: GetOptions) -> object_store::Result { + self.log(Op::Get(location.to_string())); + if self.state().hide_gets.iter().any(|p| location.as_ref().contains(p)) { + return Err(object_store::Error::NotFound { + path: location.to_string(), + source: "hidden".into(), + }); + } + self.inner.get_opts(location, options).await + } + + fn delete_stream( + &self, + locations: BoxStream<'static, object_store::Result>, + ) -> BoxStream<'static, object_store::Result> { + let state = self.state.clone(); + let locations = locations + .inspect_ok(move |path| state.lock().unwrap().ops.push(Op::Delete(path.to_string()))) + .boxed(); + self.inner.delete_stream(locations) + } + + fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result> { + self.listed(prefix, None) + } + + fn list_with_offset(&self, prefix: Option<&Path>, offset: &Path) -> BoxStream<'static, object_store::Result> { + self.listed(prefix, Some(offset)) + } + + async fn list_with_delimiter(&self, prefix: Option<&Path>) -> object_store::Result { + self.inner.list_with_delimiter(prefix).await + } + + async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> object_store::Result<()> { + self.inner.copy_opts(from, to, options).await + } +} + +/// ListObjectsV2 semantics: a raw string prefix, lexical order, an exclusive `start-after` +/// offset, at most 1000 keys, and an opaque continuation token that resumes after the last key. +#[async_trait::async_trait] +impl PaginatedListStore for Mock { + async fn list_paginated( + &self, + prefix: Option<&str>, + opts: PaginatedListOptions, + ) -> object_store::Result { + if opts.delimiter.is_some() { + return Err(unsupported("delimiter")); + } + self.log(Op::List { + prefix: prefix.unwrap_or_default().to_string(), + offset: opts.offset.clone(), + }); + + let mut metas: Vec = self.inner.list(None).try_collect().await?; + metas.sort_by(|a, b| a.location.as_ref().cmp(b.location.as_ref())); + let after = opts.page_token.or(opts.offset); + metas.retain(|meta| { + let key = meta.location.as_ref(); + prefix.is_none_or(|prefix| key.starts_with(prefix)) && after.as_deref().is_none_or(|after| key > after) + }); + + let take = opts.max_keys.unwrap_or(MAX_KEYS).clamp(1, MAX_KEYS); + let page_token = (metas.len() > take).then(|| metas[take - 1].location.to_string()); + metas.truncate(take); + Ok(PaginatedListResult { + result: ListResult { + objects: metas, + common_prefixes: Vec::new(), + extensions: Default::default(), + }, + page_token, + }) + } +} diff --git a/rs/moq-archive/src/reader/tests.rs b/rs/moq-archive/src/reader/tests.rs index ca817bf72a..b0e6d98769 100644 --- a/rs/moq-archive/src/reader/tests.rs +++ b/rs/moq-archive/src/reader/tests.rs @@ -1,89 +1,19 @@ -use std::sync::{Arc, Mutex}; - use bytes::Bytes; -use futures::stream::BoxStream; use hang::timeline::{Range, Record}; use moq_json::window; use moq_net::{broadcast, group}; -use object_store::memory::InMemory; -use object_store::path::Path; -use object_store::{ - CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, ObjectStoreExt, - PutMultipartOptions, PutOptions, PutPayload, PutResult, -}; +use object_store::ObjectStoreExt; use super::*; +use crate::mock::Mock; use crate::segment::{Frame, Group}; use crate::{ID_MAX, Info}; const TIMELINE: &str = "timeline.z"; -/// In-memory store that records every GET path. -#[derive(Debug, Clone, Default)] -struct Counting { - inner: Arc, - gets: Arc>>, -} - -impl Counting { - fn gets(&self) -> Vec { - self.gets.lock().unwrap().clone() - } -} - -impl std::fmt::Display for Counting { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "Counting") - } -} - -#[async_trait::async_trait] -impl ObjectStore for Counting { - async fn put_opts( - &self, - location: &Path, - payload: PutPayload, - opts: PutOptions, - ) -> object_store::Result { - self.inner.put_opts(location, payload, opts).await - } - - async fn put_multipart_opts( - &self, - location: &Path, - opts: PutMultipartOptions, - ) -> object_store::Result> { - self.inner.put_multipart_opts(location, opts).await - } - - async fn get_opts(&self, location: &Path, options: GetOptions) -> object_store::Result { - self.gets.lock().unwrap().push(location.to_string()); - self.inner.get_opts(location, options).await - } - - fn delete_stream( - &self, - locations: BoxStream<'static, object_store::Result>, - ) -> BoxStream<'static, object_store::Result> { - self.inner.delete_stream(locations) - } - - fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result> { - self.inner.list(prefix) - } - - async fn list_with_delimiter(&self, prefix: Option<&Path>) -> object_store::Result { - self.inner.list_with_delimiter(prefix).await - } - - async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> object_store::Result<()> { - self.inner.copy_opts(from, to, options).await - } -} - /// Writes archive objects the way a recording writer lays them out. struct Archive { - store: Store, + store: Store, encoder: window::Encoder, /// Next timeline group sequence. sequence: u64, @@ -96,7 +26,7 @@ impl Archive { /// `op_ratio` 0 makes every window edit its own checkpoint group. async fn with_op_ratio(op_ratio: u32) -> Self { - let store = Store::new(Counting::default(), "rec"); + let store = Store::new(Mock::memory(), "rec"); store.put_info(TIMELINE, &Info::new(0, 1000).unwrap()).await.unwrap(); let config = window::ProducerConfig::default() .with_compression(true) @@ -188,7 +118,7 @@ fn record(segment: u64, tracks: &[(&str, &[(u64, u64)])]) -> Record { record } -async fn open(archive: &Archive) -> (broadcast::Producer, Reader) { +async fn open(archive: &Archive) -> (broadcast::Producer, Reader) { let broadcast = broadcast::Info::new().produce(); let reader = Reader::open(archive.store.clone(), &broadcast, Config::new(TIMELINE)) .await @@ -269,7 +199,7 @@ async fn requests_download_only_their_object() { .await; let (broadcast, _reader) = open(&archive).await; - let before = archive.store.inner().gets().len(); + archive.store.inner().take(); for sequence in 0..3 { assert_eq!( @@ -278,7 +208,7 @@ async fn requests_download_only_their_object() { ); } - let gets = archive.store.inner().gets()[before..].to_vec(); + let gets = archive.store.inner().gets(); assert_eq!( gets, vec![ @@ -365,12 +295,12 @@ async fn refresh_follows_new_segments_and_pops() { ); // The popped record's groups are gone, although its object is still stored and was cached. - let gets = archive.store.inner().gets().len(); + archive.store.inner().take(); assert!(matches!( fetch(&broadcast, "video", 1, 0).await, Err(moq_net::Error::NotFound) )); - assert_eq!(archive.store.inner().gets().len(), gets); + assert_eq!(archive.store.inner().gets(), Vec::::new()); } #[tokio::test] @@ -440,7 +370,7 @@ async fn timeline_track_is_republished_and_finished_on_request() { #[tokio::test] async fn open_requires_the_timeline_info() { - let store = Store::new(Counting::default(), "rec"); + let store = Store::new(Mock::memory(), "rec"); let broadcast = broadcast::Info::new().produce(); let result = Reader::open(store, &broadcast, Config::new(TIMELINE)).await; assert!(matches!(result, Err(Error::NotFound(_)))); diff --git a/rs/moq-archive/src/store.rs b/rs/moq-archive/src/store.rs index 3a9092703e..1948d38cf0 100644 --- a/rs/moq-archive/src/store.rs +++ b/rs/moq-archive/src/store.rs @@ -343,113 +343,16 @@ mod tests { use std::num::NonZeroUsize; use futures::TryStreamExt; - use object_store::list::{PaginatedListOptions, PaginatedListResult, PaginatedListStore}; + use object_store::ObjectStoreExt; use object_store::memory::InMemory; use object_store::path::Path; - use object_store::{ - CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, PutMultipartOptions, - PutOptions, PutPayload, PutResult, - }; use super::*; use crate::ID_MAX; + use crate::mock::Mock; use crate::path::encode_track; use crate::segment::{Frame, Group}; - /// In-memory store with a trivial offset-based paginated listing implementation. - /// Page tokens are decimal indexes into the filtered, sorted key list. - #[derive(Debug, Clone)] - struct PaginatedMemory { - inner: InMemory, - } - - impl std::fmt::Display for PaginatedMemory { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "PaginatedMemory") - } - } - - #[async_trait::async_trait] - impl ObjectStore for PaginatedMemory { - async fn put_opts( - &self, - location: &Path, - payload: PutPayload, - opts: PutOptions, - ) -> object_store::Result { - self.inner.put_opts(location, payload, opts).await - } - - async fn put_multipart_opts( - &self, - location: &Path, - opts: PutMultipartOptions, - ) -> object_store::Result> { - self.inner.put_multipart_opts(location, opts).await - } - - async fn get_opts(&self, location: &Path, options: GetOptions) -> object_store::Result { - self.inner.get_opts(location, options).await - } - - fn delete_stream( - &self, - locations: BoxStream<'static, object_store::Result>, - ) -> BoxStream<'static, object_store::Result> { - self.inner.delete_stream(locations) - } - - fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result> { - self.inner.list(prefix) - } - - async fn list_with_delimiter(&self, prefix: Option<&Path>) -> object_store::Result { - self.inner.list_with_delimiter(prefix).await - } - - async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> object_store::Result<()> { - self.inner.copy_opts(from, to, options).await - } - } - - #[async_trait::async_trait] - impl PaginatedListStore for PaginatedMemory { - async fn list_paginated( - &self, - prefix: Option<&str>, - opts: PaginatedListOptions, - ) -> object_store::Result { - let mut metas: Vec = self.inner.list(None).try_collect().await?; - metas.sort_by(|a, b| a.location.as_ref().cmp(b.location.as_ref())); - if let Some(prefix) = prefix { - metas.retain(|meta| meta.location.as_ref().starts_with(prefix)); - } - if let Some(offset) = opts.offset.as_deref() { - metas.retain(|meta| meta.location.as_ref() > offset); - } - let start: usize = match opts.page_token { - Some(token) => token.parse().map_err(|err| object_store::Error::Generic { - store: "PaginatedMemory", - source: Box::new(err), - })?, - None => 0, - }; - let remaining = &metas[start.min(metas.len())..]; - let take = opts.max_keys.unwrap_or(remaining.len()).min(remaining.len()); - let objects = remaining[..take].to_vec(); - let next = start.saturating_add(take); - let page_token = (next < metas.len()).then(|| next.to_string()); - Ok(PaginatedListResult { - result: ListResult { - objects, - common_prefixes: Vec::new(), - extensions: Default::default(), - }, - page_token, - }) - } - } - fn memory() -> Store { Store::new(InMemory::new(), "rec") } @@ -818,7 +721,7 @@ mod tests { #[tokio::test] async fn paginated_listing_walks_pages_and_excludes_siblings() { - let store = Store::new(PaginatedMemory { inner: InMemory::new() }, "rec"); + let store = Store::new(Mock::memory(), "rec"); for segment in 0..3 { store .put_segments("timeline.z", segment, &one_group(segment, b"t")) @@ -827,7 +730,6 @@ mod tests { } store .inner() - .inner .put( &Path::from("rec-other/video/.info"), Bytes::from_static(b"sibling").into(), diff --git a/rs/moq-archive/src/writer.rs b/rs/moq-archive/src/writer.rs index 5f05cdc083..838561859a 100644 --- a/rs/moq-archive/src/writer.rs +++ b/rs/moq-archive/src/writer.rs @@ -803,91 +803,12 @@ fn malformed(track: &str, sequence: u64, err: impl std::fmt::Display) -> Error { mod tests { use futures::TryStreamExt; - use futures::stream::BoxStream; use object_store::memory::InMemory; - use object_store::path::Path; - use object_store::{ - CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, PutMultipartOptions, PutOptions, - PutPayload, PutResult, - }; use super::*; + use crate::mock::Mock; use crate::store::list::Query; - /// 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 { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "Failing") - } - } - - #[async_trait::async_trait] - impl ObjectStore for Failing { - async fn put_opts( - &self, - location: &Path, - payload: PutPayload, - opts: PutOptions, - ) -> object_store::Result { - if location.as_ref().contains(&format!("/{}/groups/", self.track)) { - return Err(object_store::Error::NotImplemented { - operation: "put".into(), - implementer: "Failing".into(), - }); - } - self.inner.put_opts(location, payload, opts).await - } - - async fn put_multipart_opts( - &self, - location: &Path, - opts: PutMultipartOptions, - ) -> object_store::Result> { - self.inner.put_multipart_opts(location, opts).await - } - - async fn get_opts(&self, location: &Path, options: GetOptions) -> object_store::Result { - self.inner.get_opts(location, options).await - } - - fn delete_stream( - &self, - locations: BoxStream<'static, object_store::Result>, - ) -> BoxStream<'static, object_store::Result> { - self.inner.delete_stream(locations) - } - - fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result> { - 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 { - self.inner.list_with_delimiter(prefix).await - } - - async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> object_store::Result<()> { - self.inner.copy_opts(from, to, options).await - } - } - const TIMELINE: &str = hang::timeline::DEFAULT_NAME; fn ms(v: u64) -> Timestamp { @@ -1031,12 +952,9 @@ mod tests { let video = track(&source, "video"); let audio = track(&source, "audio"); - let failing = Failing { - inner: Arc::new(InMemory::new()), - track: "audio", - list: false, - }; - let store = Store::new(failing, "rec"); + let mock = Mock::memory(); + mock.fail_puts("/audio/groups/"); + let store = Store::new(mock, "rec"); let writer = Writer::new(store.clone(), source.consume(), Config::default()) .await .unwrap(); @@ -1367,18 +1285,15 @@ mod tests { #[tokio::test] async fn a_failed_recovery_deletes_nothing() { - let inner = Arc::new(InMemory::new()); - let store = Store::new(inner.clone(), "rec"); + let mock = Mock::memory(); + let store = Store::new(mock.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 failing = mock.fork(); + failing.fail_lists(); 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(_)))); From 90d68d0e7427c0d8b68598bca7330d7875e5e635 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:21:17 -0700 Subject: [PATCH 03/10] test(archive): cover key, table, and listing edge cases Co-Authored-By: Claude Opus 5.5 --- rs/moq-archive/src/path.rs | 47 ++++++++++++++++++ rs/moq-archive/src/segment.rs | 54 ++++++++++++++++++++ rs/moq-archive/src/store.rs | 93 +++++++++++++++++++++++++++++++++++ 3 files changed, 194 insertions(+) diff --git a/rs/moq-archive/src/path.rs b/rs/moq-archive/src/path.rs index 86a8294e7b..0ed2ec682b 100644 --- a/rs/moq-archive/src/path.rs +++ b/rs/moq-archive/src/path.rs @@ -280,6 +280,10 @@ mod tests { parse_id("0009007199254740992"), Err(Error::Id(n)) if n == ID_MAX + 1 )); + // The largest QUIC varint still fits the 19-digit field. + let varint = (1u64 << 62) - 1; + assert!(matches!(format_id(varint), Err(Error::Id(n)) if n == varint)); + assert!(matches!(parse_id("4611686018427387903"), Err(Error::Id(n)) if n == varint)); assert!(matches!(parse_id("5"), Err(Error::Path(_)))); assert!(matches!(parse_id("000000000000000000X"), Err(Error::Path(_)))); } @@ -334,6 +338,49 @@ mod tests { assert!(matches!(Key::groups("v", ID_MAX + 1..=ID_MAX + 1), Err(Error::Id(_)))); } + #[test] + fn direct_construction_is_validated_when_serialized() { + let prefix = Path::from("rec"); + let empty = Key::Info { track: String::new() }; + assert!(matches!(empty.path(&prefix), Err(Error::Track))); + let segment = Key::Segments { + track: "t".to_string(), + segment: ID_MAX + 1, + }; + assert!(matches!(segment.path(&prefix), Err(Error::Id(_)))); + let varint = Key::Groups { + track: "v".to_string(), + range: 0..=(1 << 62) - 1, + }; + assert!(matches!(varint.path(&prefix), Err(Error::Id(_)))); + + // Everything that does serialize parses back to the same key. + for key in [ + Key::Info { + track: "a/b".to_string(), + }, + Key::Groups { + track: ".x".to_string(), + range: ID_MAX..=ID_MAX, + }, + Key::Segments { + track: "é".to_string(), + segment: 0, + }, + ] { + let path = key.path(&prefix).unwrap(); + assert_eq!(Key::parse(&prefix, &path).unwrap(), key); + } + } + + #[test] + fn keys_parse_under_an_empty_prefix() { + let key = Key::groups("catalog.json", 1..=2).unwrap(); + let path = key.path(&Path::ROOT).unwrap(); + assert_eq!(path.as_ref(), "catalog%2Ejson/groups/0000000000000000002.0000000000000000001"); + assert_eq!(Key::parse(&Path::ROOT, &path).unwrap(), key); + } + #[test] fn direct_inverted_range_is_rejected_when_serialized() { let (smallest, largest) = (2, 1); diff --git a/rs/moq-archive/src/segment.rs b/rs/moq-archive/src/segment.rs index c2fd68c5aa..f2aa3fadd2 100644 --- a/rs/moq-archive/src/segment.rs +++ b/rs/moq-archive/src/segment.rs @@ -346,6 +346,35 @@ mod tests { assert!(matches!(Object::decode(bytes.freeze()), Err(Error::Id(_)))); } + /// A table with one frameless group per delta. + fn deltas(deltas: &[u64]) -> Bytes { + let mut bytes = BytesMut::new(); + write_varint(&mut bytes, 1).unwrap(); + write_varint(&mut bytes, deltas.len() as u64).unwrap(); + for &delta in deltas { + write_varint(&mut bytes, delta).unwrap(); + write_varint(&mut bytes, 0).unwrap(); + } + bytes.freeze() + } + + fn sequences(bytes: Bytes) -> Result> { + Ok(Object::decode(bytes)?.groups.iter().map(|group| group.sequence).collect()) + } + + #[test] + fn delta_extremes() { + let varint = (1u64 << 62) - 1; + assert_eq!(sequences(deltas(&[0, 0, 0])).unwrap(), vec![0, 1, 2]); + assert_eq!(sequences(deltas(&[0, 9, 0])).unwrap(), vec![0, 10, 11]); + assert_eq!(sequences(deltas(&[0, ID_MAX - 1])).unwrap(), vec![0, ID_MAX]); + assert_eq!(sequences(deltas(&[ID_MAX])).unwrap(), vec![ID_MAX]); + assert!(matches!(sequences(deltas(&[0, ID_MAX])), Err(Error::Id(_)))); + assert!(matches!(sequences(deltas(&[ID_MAX + 1])), Err(Error::Id(_)))); + assert!(matches!(sequences(deltas(&[varint])), Err(Error::Id(_)))); + assert!(matches!(sequences(deltas(&[ID_MAX, varint])), Err(Error::Id(_)))); + } + #[test] fn timestamp_bounds() { assert!( @@ -413,6 +442,31 @@ mod tests { assert!(matches!(Object::decode(&b"\x01"[..]), Err(Error::Table))); } + #[test] + fn frame_offsets_must_tile_the_payload() { + // One group of two frames at (offset, length), followed by `payload` bytes. + let table = |frames: [(u64, u64); 2], payload: usize| { + let mut bytes = BytesMut::new(); + for value in [1, 1, 0, 2] { + write_varint(&mut bytes, value).unwrap(); + } + for (offset, length) in frames { + for value in [0, offset, length] { + write_varint(&mut bytes, value).unwrap(); + } + } + bytes.extend_from_slice(&vec![b'x'; payload]); + Object::decode(bytes.freeze()) + }; + assert!(table([(0, 1), (1, 2)], 3).is_ok()); + assert!(matches!(table([(0, 1), (0, 2)], 3), Err(Error::Table)), "overlap"); + assert!(matches!(table([(0, 1), (2, 1)], 3), Err(Error::Table)), "gap"); + assert!(matches!(table([(1, 1), (2, 1)], 3), Err(Error::Table)), "late start"); + assert!(matches!(table([(0, 1), (1, 3)], 3), Err(Error::Table)), "past the end"); + let varint = (1u64 << 62) - 1; + assert!(matches!(table([(0, 1), (1, varint)], 3), Err(Error::Table)), "huge length"); + } + #[test] fn filename_bounds_must_match_the_table() { let object = object(vec![ diff --git a/rs/moq-archive/src/store.rs b/rs/moq-archive/src/store.rs index 1948d38cf0..63cf2ec709 100644 --- a/rs/moq-archive/src/store.rs +++ b/rs/moq-archive/src/store.rs @@ -423,6 +423,30 @@ mod tests { assert_eq!(&kept[..], br#"{ "timescale": 1000, "priority": 1, "version": 1 }"#); } + #[tokio::test] + async fn malformed_or_unsupported_existing_info_is_refused_and_kept() { + let store = memory(); + let info = Info::new(0, 1_000).unwrap(); + for (track, existing, check) in [ + ( + "v2", + &br#"{"version":2,"priority":0,"timescale":1000}"#[..], + (|err| matches!(err, Error::Version(2))) as fn(&Error) -> bool, + ), + ("junk", b"not json", |err| matches!(err, Error::Json(_))), + ("zero", br#"{"version":1,"priority":0,"timescale":0}"#, |err| { + matches!(err, Error::Timescale(0)) + }), + ] { + let path = store.path(&Key::info(track).unwrap()).unwrap(); + store.inner().put(&path, existing.to_vec().into()).await.unwrap(); + let err = store.put_info(track, &info).await.unwrap_err(); + assert!(check(&err), "{track}: {err}"); + let kept = store.inner().get(&path).await.unwrap().bytes().await.unwrap(); + assert_eq!(&kept[..], existing, "{track} is not rewritten"); + } + } + #[tokio::test] async fn info_property_mismatch_is_a_hard_error() { let store = memory(); @@ -798,6 +822,75 @@ mod tests { assert_eq!(paged, streamed); } + /// The first object listed after `groups_from(group)`, one S3 page of one key. + async fn lookup(store: &Store, group: u64) -> Option> { + let query = Query::groups_from("video", group) + .unwrap() + .page_size(NonZeroUsize::new(1).unwrap()); + let page = store.list_paginated(&query).await.unwrap(); + match page.entries.first().map(|entry| &entry.key) { + Some(Key::Groups { range, .. }) => Some(range.clone()), + Some(key) => panic!("unexpected {key:?}"), + None => None, + } + } + + #[tokio::test] + async fn ordered_lookup_finds_the_covering_object() { + let store = Store::new(Mock::memory(), "rec"); + for range in [0..=2, 5..=7, 10..=10, ID_MAX..=ID_MAX] { + let groups = range.clone().filter(|sequence| *sequence != 6); + let object = Object { + groups: groups + .map(|sequence| Group { + sequence, + frames: vec![frame(0, b"g")], + }) + .collect(), + }; + store.put_groups("video", &object).await.unwrap(); + } + // A sibling track sorts after `video/groups/` and must never be returned. + store.put_groups("video-alt", &one_group(3, b"a")).await.unwrap(); + + // Largest-first filenames make the first key at or past the group its only candidate. + for (group, found) in [ + (0, Some(0..=2)), + (1, Some(0..=2)), + (2, Some(0..=2)), + (3, Some(5..=7)), + (5, Some(5..=7)), + (6, Some(5..=7)), + (7, Some(5..=7)), + (8, Some(10..=10)), + (10, Some(10..=10)), + (11, Some(ID_MAX..=ID_MAX)), + (ID_MAX, Some(ID_MAX..=ID_MAX)), + ] { + assert_eq!(lookup(&store, group).await, found, "group {group}"); + } + store.delete(&Key::groups("video", ID_MAX..=ID_MAX).unwrap()).await.unwrap(); + assert_eq!(lookup(&store, 11).await, None); + assert!(Query::groups_from("video", ID_MAX + 1).is_err()); + } + + #[tokio::test] + async fn empty_prefix_lists_the_whole_store() { + let store = Store::new(Mock::memory(), ""); + store.put_info("catalog.json", &Info::new(0, 1).unwrap()).await.unwrap(); + store.put_groups("video", &one_group(4, b"a")).await.unwrap(); + assert_eq!(store.paginated_prefix(None), None); + + let expected = std::collections::HashSet::from([Key::info("catalog.json").unwrap(), Key::groups("video", 4..=4).unwrap()]); + let streamed: std::collections::HashSet = store.list(&Query::new()).map_ok(|entry| entry.key).try_collect().await.unwrap(); + assert_eq!(streamed, expected); + let page = store.list_paginated(&Query::new()).await.unwrap(); + assert_eq!(page.entries.into_iter().map(|entry| entry.key).collect::>(), expected); + assert!(page.next.is_none()); + let page = store.list_paginated(&Query::groups("video").unwrap()).await.unwrap(); + assert_eq!(page.entries.len(), 1); + } + #[test] fn directory_results_are_not_silently_dropped() { let store = memory(); From 330bedcf1fa8d304db2737024248e3140ad19eb5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:30:25 -0700 Subject: [PATCH 04/10] test(archive): pin reader following, missing tails, and unusable .info Co-Authored-By: Claude Opus 5.5 --- rs/moq-archive/src/reader/tests.rs | 148 ++++++++++++++++++++++++++++- 1 file changed, 147 insertions(+), 1 deletion(-) diff --git a/rs/moq-archive/src/reader/tests.rs b/rs/moq-archive/src/reader/tests.rs index b0e6d98769..723f02dd29 100644 --- a/rs/moq-archive/src/reader/tests.rs +++ b/rs/moq-archive/src/reader/tests.rs @@ -5,7 +5,7 @@ use moq_net::{broadcast, group}; use object_store::ObjectStoreExt; use super::*; -use crate::mock::Mock; +use crate::mock::{Mock, Op}; use crate::segment::{Frame, Group}; use crate::{ID_MAX, Info}; @@ -326,6 +326,152 @@ async fn a_missing_timeline_segment_recovers_from_the_next_checkpoint() { } } +#[tokio::test] +async fn a_missing_tail_is_retried_on_the_next_refresh() { + let mut archive = Archive::new().await; + for segment in 0..3 { + archive.media("video", &[(segment, 1)]).await; + archive.commit(&record(segment, &[("video", &[(segment, segment)])]), 0).await; + } + // Listed, but not yet readable. + archive.store.inner().hide_gets("segments/0000000000000000002"); + + let (broadcast, mut reader) = open(&archive).await; + assert!(fetch(&broadcast, "video", 1, 0).await.is_ok()); + assert!(matches!( + fetch(&broadcast, "video", 2, 0).await, + Err(moq_net::Error::NotFound) + )); + + archive.store.inner().heal(); + archive.store.inner().take(); + reader.refresh().await.unwrap(); + assert_eq!( + archive.store.inner().take(), + [ + Op::List { + prefix: "rec/timeline%2Ez/segments".to_string(), + offset: Some("rec/timeline%2Ez/segments/0000000000000000001".to_string()), + }, + Op::Get("rec/timeline%2Ez/segments/0000000000000000002".to_string()), + ], + "the cursor stays before the missing tail" + ); + assert_eq!( + fetch(&broadcast, "video", 2, 0).await.unwrap(), + expected("video", 2, 0..1) + ); +} + +#[tokio::test] +async fn following_lists_only_new_timeline_keys() { + let mut archive = Archive::new().await; + archive.media("video", &[(0, 1)]).await; + archive.commit(&record(0, &[("video", &[(0, 0)])]), 0).await; + let (broadcast, mut reader) = open(&archive).await; + + for segment in 1..4 { + // A media object stored ahead of its commit is invisible until the timeline names it. + archive.media("video", &[(segment, 1)]).await; + reader.refresh().await.unwrap(); + assert!(matches!( + fetch(&broadcast, "video", segment, 0).await, + Err(moq_net::Error::NotFound) + )); + + archive.commit(&record(segment, &[("video", &[(segment, segment)])]), 0).await; + archive.store.inner().take(); + reader.refresh().await.unwrap(); + let previous = format!("rec/timeline%2Ez/segments/{:019}", segment - 1); + assert_eq!( + archive.store.inner().take(), + [ + Op::List { + prefix: "rec/timeline%2Ez/segments".to_string(), + offset: Some(previous), + }, + Op::Get(format!("rec/timeline%2Ez/segments/{segment:019}")), + ], + "following segment {segment} touches no media listing" + ); + assert_eq!( + fetch(&broadcast, "video", segment, 0).await.unwrap(), + expected("video", segment, 0..1) + ); + } +} + +#[tokio::test] +async fn an_unordered_listing_replays_in_segment_order() { + let mut archive = Archive::new().await; + archive.store.inner().unordered(); + for segment in 0..4 { + archive.media("video", &[(segment, 1)]).await; + let pop = u64::from(segment >= 2); + archive.commit(&record(segment, &[("video", &[(segment, segment)])]), pop).await; + } + + let (broadcast, mut reader) = open(&archive).await; + + // The cursor is the newest segment, not the last one listed. + archive.media("video", &[(4, 1)]).await; + archive.commit(&record(4, &[("video", &[(4, 4)])]), 1).await; + archive.store.inner().take(); + reader.refresh().await.unwrap(); + assert_eq!( + archive.store.inner().gets(), + ["rec/timeline%2Ez/segments/0000000000000000004"] + ); + + for segment in 0..3 { + assert!(matches!( + fetch(&broadcast, "video", segment, 0).await, + Err(moq_net::Error::NotFound) + )); + } + for segment in 3..5 { + assert_eq!( + fetch(&broadcast, "video", segment, 0).await.unwrap(), + expected("video", segment, 0..1) + ); + } +} + +#[tokio::test] +async fn a_track_without_usable_info_is_not_found() { + let mut archive = Archive::new().await; + archive.media("audio", &[(0, 1)]).await; + let object = |track| Object { + groups: vec![Group { + sequence: 0, + frames: vec![frame(track, 0, 0)], + }], + }; + archive.store.put_groups("bare", &object("bare")).await.unwrap(); + archive.store.put_groups("future", &object("future")).await.unwrap(); + archive + .raw(&Key::info("future").unwrap(), br#"{"version":2,"priority":0,"timescale":1000}"#) + .await; + archive + .commit( + &record(0, &[("audio", &[(0, 0)]), ("bare", &[(0, 0)]), ("future", &[(0, 0)])]), + 0, + ) + .await; + + let (broadcast, _reader) = open(&archive).await; + for track in ["bare", "future"] { + assert!( + matches!(fetch(&broadcast, track, 0, 0).await, Err(moq_net::Error::NotFound)), + "{track}" + ); + } + assert_eq!( + fetch(&broadcast, "audio", 0, 0).await.unwrap(), + expected("audio", 0, 0..1) + ); +} + #[tokio::test] async fn multi_frame_timeline_groups_decode() { let mut archive = Archive::with_op_ratio(64).await; From 1f50109de9605fce1022257a551500b4e7131fe5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:32:46 -0700 Subject: [PATCH 05/10] test(archive): prove DVR crash cleanup, archive orphans, and out-of-order completion Co-Authored-By: Claude Opus 5.5 --- rs/moq-archive/src/writer.rs | 113 +++++++++++++++++++++++++++++++++++ 1 file changed, 113 insertions(+) diff --git a/rs/moq-archive/src/writer.rs b/rs/moq-archive/src/writer.rs index 838561859a..a46d77eac1 100644 --- a/rs/moq-archive/src/writer.rs +++ b/rs/moq-archive/src/writer.rs @@ -1043,6 +1043,39 @@ mod tests { run.abort(); } + #[tokio::test] + async fn accepted_groups_may_complete_out_of_order() { + 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(); + writer.control().pacing_track("video").await.unwrap(); + let run = tokio::spawn(writer.run()); + + let mut first = video.create_group(group::Info { sequence: 0 }).unwrap(); + first.write_frame(ms(0), "0@0").unwrap(); + group(&video, 1, &[1000]); + group(&video, 2, &[2000]); + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(window(&store).await.is_empty(), "group 1 waits for group 0"); + + first.write_frame(ms(500), "0@500").unwrap(); + first.finish().unwrap(); + video.finish().unwrap(); + source.finish(); + run.await.unwrap().unwrap(); + + let records = window(&store).await; + assert_eq!(ranges(&records, "video"), vec![(0, 0), (1, 1), (2, 2)]); + check_objects(&store, &records).await; + let object = store.get_groups("video", 0..=0).await.unwrap(); + let frames: Vec<_> = object.groups[0].frames.iter().map(|f| f.timestamp).collect(); + assert_eq!(frames, vec![0, 500]); + } + #[tokio::test] async fn decreasing_arrivals_are_refused() { let source = broadcast::Info::new().produce(); @@ -1283,6 +1316,86 @@ mod tests { store.get_segments(TIMELINE, 0).await.unwrap(); } + #[tokio::test] + async fn a_dvr_crash_between_pop_and_delete_is_cleaned_on_restart() { + let store = Store::new(InMemory::new(), "rec"); + + // A grace longer than the test keeps every expired object past the crash. + let source = broadcast::Info::new().produce(); + let video = track(&source, "video"); + let config = + Config::default().with_retention(Retention::new(Duration::from_secs(2), Duration::from_secs(3600))); + let writer = Writer::new(store.clone(), source.consume(), config).await.unwrap(); + writer.control().pacing_track("video").await.unwrap(); + let run = tokio::spawn(writer.run()); + for sequence in 0..6 { + group(&video, sequence, &[sequence * 1000, sequence * 1000 + 500]); + } + // Segment 5 stays open, so the newest durable timeline object is segment 4. + while window(&store).await.last().map(|record| record.segment) != Some(4) { + tokio::time::sleep(Duration::from_millis(10)).await; + } + run.abort(); + let _ = run.await; + + let retained = window(&store).await; + assert_eq!( + retained.iter().map(|record| record.segment).collect::>(), + vec![3, 4] + ); + let expired: HashSet<_> = (0..3).map(|s| Key::groups("video", s..=s).unwrap()).collect(); + assert_eq!(stored_groups(&store).await, &referenced(&retained) | &expired); + // An upload the crash left uncommitted. + store.put_groups("video", &orphan(5)).await.unwrap(); + + let grace = Duration::from_millis(200); + let config = Config::default().with_retention(Retention::new(Duration::from_secs(2), grace)); + let source = broadcast::Info::new().produce(); + let video = track(&source, "video"); + let started = Instant::now(); + let writer = Writer::new(store.clone(), source.consume(), config).await.unwrap(); + assert!( + stored_groups(&store).await.is_superset(&expired), + "expired objects outlive the grace, for readers holding the old timeline" + ); + + writer.control().pacing_track("video").await.unwrap(); + // The source replays the uncommitted group; it is refused rather than overwritten. + for sequence in 5..9 { + group(&video, sequence, &[sequence * 1000, sequence * 1000 + 500]); + } + video.finish().unwrap(); + source.finish(); + writer.run().await.unwrap(); + assert!(started.elapsed() >= grace); + + let records = window(&store).await; + assert_eq!(ranges(&records, "video"), vec![(6, 6), (7, 7), (8, 8)]); + 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(); + for segment in 0..=7 { + store.get_segments(TIMELINE, segment).await.unwrap(); + } + } + + #[tokio::test] + async fn an_archive_restart_leaves_uncommitted_groups_unadvertised() { + let store = Store::new(InMemory::new(), "rec"); + record(&store, Config::default(), 0..3).await; + // Media stored after the last timeline commit: a crash before its record. + store.put_groups("video", &orphan(3)).await.unwrap(); + + record(&store, Config::default(), 3..6).await; + let records = window(&store).await; + assert_eq!(ranges(&records, "video"), vec![(0, 0), (1, 1), (2, 2), (4, 4), (5, 5)]); + check_objects(&store, &records).await; + // An archive deletes nothing; the orphan stays invisible. + assert!(stored_groups(&store).await.contains(&Key::groups("video", 3..=3).unwrap())); + assert!(!referenced(&records).contains(&Key::groups("video", 3..=3).unwrap())); + } + #[tokio::test] async fn a_failed_recovery_deletes_nothing() { let mock = Mock::memory(); From 369470f1098b6da69c94e8abad3d65d294f00658 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:41:06 -0700 Subject: [PATCH 06/10] fix(archive): stamp stored timeline frames with content time The live timeline track stamps frames with the wall clock, so two recordings of the same content produced different timeline objects. Store each segment's timeline groups at the record's pts instead, and prove the full recording is byte-identical across memory, local, and unordered-listing backends, replays exactly through FETCH, and reads only the requested rendition. Co-Authored-By: Claude Opus 5.5 --- rs/moq-archive/src/lib.rs | 2 + rs/moq-archive/src/mock.rs | 24 +- rs/moq-archive/src/path.rs | 5 +- rs/moq-archive/src/proof.rs | 395 +++++++++++++++++++++++++++++ rs/moq-archive/src/reader/tests.rs | 17 +- rs/moq-archive/src/segment.rs | 11 +- rs/moq-archive/src/store.rs | 23 +- rs/moq-archive/src/writer.rs | 25 +- 8 files changed, 482 insertions(+), 20 deletions(-) create mode 100644 rs/moq-archive/src/proof.rs diff --git a/rs/moq-archive/src/lib.rs b/rs/moq-archive/src/lib.rs index f84d4c8a01..3edecbe87e 100644 --- a/rs/moq-archive/src/lib.rs +++ b/rs/moq-archive/src/lib.rs @@ -20,6 +20,8 @@ pub mod info; #[cfg(test)] mod mock; mod path; +#[cfg(test)] +mod proof; pub mod reader; mod recover; pub mod segment; diff --git a/rs/moq-archive/src/mock.rs b/rs/moq-archive/src/mock.rs index 1869b1455c..2d7275badf 100644 --- a/rs/moq-archive/src/mock.rs +++ b/rs/moq-archive/src/mock.rs @@ -23,7 +23,10 @@ pub(crate) enum Op { Put(String), Delete(String), /// A streaming or paginated listing, with its prefix and exclusive offset. - List { prefix: String, offset: Option }, + List { + prefix: String, + offset: Option, + }, } #[derive(Debug, Default)] @@ -117,7 +120,11 @@ impl Mock { self.state().ops.push(op); } - fn listed(&self, prefix: Option<&Path>, offset: Option<&Path>) -> BoxStream<'static, object_store::Result> { + fn listed( + &self, + prefix: Option<&Path>, + offset: Option<&Path>, + ) -> BoxStream<'static, object_store::Result> { self.log(Op::List { prefix: prefix.map(ToString::to_string).unwrap_or_default(), offset: offset.map(ToString::to_string), @@ -161,7 +168,12 @@ impl std::fmt::Display for Mock { #[async_trait::async_trait] impl ObjectStore for Mock { - async fn put_opts(&self, location: &Path, payload: PutPayload, opts: PutOptions) -> object_store::Result { + async fn put_opts( + &self, + location: &Path, + payload: PutPayload, + opts: PutOptions, + ) -> object_store::Result { self.log(Op::Put(location.to_string())); if self.state().fail_puts.iter().any(|p| location.as_ref().contains(p)) { return Err(unsupported("put")); @@ -203,7 +215,11 @@ impl ObjectStore for Mock { self.listed(prefix, None) } - fn list_with_offset(&self, prefix: Option<&Path>, offset: &Path) -> BoxStream<'static, object_store::Result> { + fn list_with_offset( + &self, + prefix: Option<&Path>, + offset: &Path, + ) -> BoxStream<'static, object_store::Result> { self.listed(prefix, Some(offset)) } diff --git a/rs/moq-archive/src/path.rs b/rs/moq-archive/src/path.rs index 0ed2ec682b..4df4784b40 100644 --- a/rs/moq-archive/src/path.rs +++ b/rs/moq-archive/src/path.rs @@ -377,7 +377,10 @@ mod tests { fn keys_parse_under_an_empty_prefix() { let key = Key::groups("catalog.json", 1..=2).unwrap(); let path = key.path(&Path::ROOT).unwrap(); - assert_eq!(path.as_ref(), "catalog%2Ejson/groups/0000000000000000002.0000000000000000001"); + assert_eq!( + path.as_ref(), + "catalog%2Ejson/groups/0000000000000000002.0000000000000000001" + ); assert_eq!(Key::parse(&Path::ROOT, &path).unwrap(), key); } diff --git a/rs/moq-archive/src/proof.rs b/rs/moq-archive/src/proof.rs new file mode 100644 index 0000000000..4acd6238ab --- /dev/null +++ b/rs/moq-archive/src/proof.rs @@ -0,0 +1,395 @@ +//! End to end: one multi-rendition broadcast recorded by a [`Writer`], stored byte-identically on +//! every backend, and replayed exactly through a [`Reader`]. + +use std::collections::BTreeMap; +use std::time::Duration; + +use bytes::Bytes; +use futures::TryStreamExt; +use moq_net::{Timescale, Timestamp, broadcast, group, track}; +use object_store::local::LocalFileSystem; +use object_store::path::Path; +use object_store::{ObjectStore, ObjectStoreExt}; + +use crate::mock::{Mock, Op}; +use crate::reader::Config as ReaderConfig; +use crate::writer::{Config, Retention}; +use crate::{Reader, Store, Writer}; + +const TIMELINE: &str = hang::timeline::DEFAULT_NAME; +const RENDITIONS: [&str; 2] = ["video/1080p", "video/360p"]; +const PACING: [&str; 3] = ["video/1080p", "video/360p", "audio"]; +const TRACKS: [&str; 5] = ["video/1080p", "video/360p", "audio", "catalog.json", "chat"]; + +/// Frames of one source group: (millisecond timestamp, payload). +type Frames = Vec<(u64, Bytes)>; + +/// Recorded segments: three 2s segments, then a one-frame tail. +const SEGMENTS: u64 = 4; + +/// The source groups of one segment: (track, sequence, frames). +/// +/// Video renditions share a 2s GOP and audio has four groups per segment. The catalog and a +/// non-media chat track publish sparsely, including a sequence skip inside one segment and an +/// empty payload. The tail keeps any single track from cutting the final segment once the others +/// end. +fn plan(segment: u64) -> Vec<(&'static str, u64, Frames)> { + let pts = segment * 2000; + let (frames, audio) = if segment + 1 < SEGMENTS { (4, 4) } else { (1, 1) }; + let mut groups = Vec::new(); + for track in RENDITIONS { + let frames = (0..frames) + .map(|i| (pts + i * 500, Bytes::from(format!("{track} {segment}.{i}")))) + .collect(); + groups.push((track, segment, frames)); + } + for sequence in segment * 4..segment * 4 + audio { + let frames = (0..2) + .map(|i| (sequence * 500 + i * 250, Bytes::from(format!("audio {sequence}.{i}")))) + .collect(); + groups.push(("audio", sequence, frames)); + } + let catalog = |sequence: u64| Bytes::from(format!(r#"{{"version":{sequence}}}"#)); + let chat = |text: &'static str| Bytes::from_static(text.as_bytes()); + match segment { + 0 => { + groups.push(("catalog.json", 0, vec![(0, catalog(0))])); + groups.push(("chat", 0, vec![(1200, chat("hello")), (1300, Bytes::new())])); + } + 2 => { + groups.push(("catalog.json", 1, vec![(4000, catalog(1))])); + groups.push(("chat", 3, vec![(4100, chat("skip"))])); + groups.push(("chat", 5, vec![(5100, chat("bye"))])); + } + _ => {} + } + groups +} + +fn ms(value: u64) -> Timestamp { + Timestamp::new(value, Timescale::MILLI).unwrap() +} + +fn write(track: &track::Producer, sequence: u64, frames: &Frames) { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + for (timestamp, payload) in frames { + group.write_frame(ms(*timestamp), payload.clone()).unwrap(); + } + group.finish().unwrap(); +} + +/// Let the writer read and report everything already published. Reads never touch the store. +async fn settle() { + for _ in 0..256 { + tokio::task::yield_now().await; + } +} + +/// Record every segment of [`plan`] into `store`. +async fn record(store: &Store) { + let source = broadcast::Info::new().produce(); + let tracks: BTreeMap<&str, track::Producer> = TRACKS + .iter() + .map(|&name| { + let info = track::Info::default() + .with_timescale(Timescale::MILLI) + .with_max_age(Duration::from_secs(3600)); + (name, source.create_track(name, info).unwrap()) + }) + .collect(); + + let writer = Writer::new(store.clone(), source.consume(), Config::default()) + .await + .unwrap(); + let control = writer.control(); + for name in TRACKS { + match PACING.contains(&name) { + true => control.pacing_track(name).await.unwrap(), + false => control.track(name).await.unwrap(), + } + } + let run = tokio::spawn(writer.run()); + + // A non-pacing track joins whichever segment is open when its group arrives, so publish each + // segment's pacing groups (closing the previous segment) before its sparse groups. + for segment in 0..SEGMENTS { + let (pacing, sparse): (Vec<_>, Vec<_>) = plan(segment) + .into_iter() + .partition(|(name, _, _)| PACING.contains(name)); + for (name, sequence, frames) in pacing { + write(&tracks[name], sequence, &frames); + } + settle().await; + for (name, sequence, frames) in sparse { + write(&tracks[name], sequence, &frames); + } + settle().await; + } + for track in tracks.values() { + track.finish().unwrap(); + } + source.finish(); + run.await.unwrap().unwrap(); +} + +/// Every object under the store's prefix, by path relative to the store root. +async fn objects(store: &Store) -> BTreeMap { + let metas: Vec<_> = store.inner().list(Some(store.prefix())).try_collect().await.unwrap(); + let mut objects = BTreeMap::new(); + for meta in metas { + let bytes = store.inner().get(&meta.location).await.unwrap().bytes().await.unwrap(); + objects.insert(meta.location.to_string(), bytes); + } + objects +} + +fn id(value: u64) -> String { + format!("{value:019}") +} + +/// The exact layout [`plan`] produces: no manifest, index, or completion marker. +fn layout() -> Vec { + let mut keys = Vec::new(); + let groups = |track: &str, ranges: &[(u64, u64)]| { + ranges + .iter() + .map(|&(smallest, largest)| format!("rec/{track}/groups/{}.{}", id(largest), id(smallest))) + .collect::>() + }; + keys.push("rec/audio/.info".to_string()); + keys.extend(groups("audio", &[(0, 3), (4, 7), (8, 11), (12, 12)])); + keys.push("rec/catalog%2Ejson/.info".to_string()); + keys.extend(groups("catalog%2Ejson", &[(0, 0), (1, 1)])); + keys.push("rec/chat/.info".to_string()); + keys.extend(groups("chat", &[(0, 0), (3, 5)])); + keys.push("rec/timeline%2Ez/.info".to_string()); + keys.extend((0..SEGMENTS).map(|segment| format!("rec/timeline%2Ez/segments/{}", id(segment)))); + for track in ["video%2F1080p", "video%2F360p"] { + keys.push(format!("rec/{track}/.info")); + keys.extend(groups(track, &[(0, 0), (1, 1), (2, 2), (3, 3)])); + } + keys.sort(); + keys +} + +/// FETCH one whole group, returning its frames. +async fn fetch(broadcast: &broadcast::Producer, track: &str, sequence: u64) -> moq_net::Result { + let track = broadcast.consume().track(track)?; + let mut group = track.fetch_group(sequence, group::Fetch::default()).await?; + let mut frames = Vec::new(); + while let Some(frame) = group.read_frame().await? { + frames.push(( + frame.timestamp.convert(Timescale::MILLI).unwrap().value(), + frame.payload, + )); + } + Ok(frames) +} + +#[tokio::test] +async fn recordings_are_byte_identical_on_every_backend() { + let memory = Store::new(Mock::memory(), "rec"); + record(&memory).await; + let expected = objects(&memory).await; + assert_eq!(expected.keys().cloned().collect::>(), layout()); + assert_eq!( + &expected["rec/video%2F360p/.info"][..], + br#"{"version":1,"priority":0,"timescale":1000}"# + ); + + let dir = tempfile::tempdir().unwrap(); + let local = Store::new(Mock::new(LocalFileSystem::new_with_prefix(dir.path()).unwrap()), "rec"); + record(&local).await; + assert_eq!(objects(&local).await, expected, "local disk"); + + // A backend with no listing order, and a sibling recording sharing the prefix's stem. + let unordered = Mock::memory(); + unordered.unordered(); + let sibling = Store::new(unordered.clone(), "rec-other"); + record(&sibling).await; + let unordered = Store::new(unordered, "rec"); + record(&unordered).await; + assert_eq!(objects(&unordered).await, expected, "unordered listing"); + assert_eq!(objects(&sibling).await.len(), expected.len()); + + // Every group object holds exactly the source groups, in order, byte for byte. + let mut stored: BTreeMap<(String, u64), Frames> = BTreeMap::new(); + for (path, bytes) in &expected { + let key = crate::Key::parse(&Path::from("rec"), &Path::parse(path).unwrap()).unwrap(); + if let crate::Key::Groups { track, range } = key { + let object = crate::Object::decode_groups(bytes.clone(), range).unwrap(); + for group in object.groups { + let frames = group.frames.into_iter().map(|f| (f.timestamp, f.payload)).collect(); + stored.insert((track.clone(), group.sequence), frames); + } + } + } + let source: BTreeMap<(String, u64), Frames> = (0..SEGMENTS) + .flat_map(plan) + .map(|(track, sequence, frames)| ((track.to_string(), sequence), frames)) + .collect(); + assert_eq!(stored, source); +} + +#[tokio::test] +async fn fetch_replays_the_recording_and_reads_only_the_requested_rendition() { + let mock = Mock::memory(); + let store = Store::new(mock.clone(), "rec"); + record(&store).await; + + let broadcast = broadcast::Info::new().produce(); + let reader = Reader::open(store, &broadcast, ReaderConfig::new(TIMELINE)) + .await + .unwrap(); + tokio::spawn(reader.serve()); + mock.take(); + + // Low-rendition playback never downloads the 1080p object. + for sequence in 0..SEGMENTS { + let expected = plan(sequence) + .into_iter() + .find(|(name, ..)| *name == "video/360p") + .unwrap() + .2; + assert_eq!(fetch(&broadcast, "video/360p", sequence).await.unwrap(), expected); + } + let object = + |track: &str, smallest: u64, largest: u64| format!("rec/{track}/groups/{}.{}", id(largest), id(smallest)); + // Each track request also GETs that track's `.info`, but nothing of another track. + let media = |gets: Vec, track: &str| { + let prefix = format!("rec/{track}/"); + assert!(gets.iter().all(|path| path.starts_with(&prefix)), "{gets:?}"); + gets.into_iter() + .filter(|path| path.contains("/groups/")) + .collect::>() + }; + assert_eq!( + media(mock.gets(), "video%2F360p"), + [ + object("video%2F360p", 0, 0), + object("video%2F360p", 1, 1), + object("video%2F360p", 2, 2), + object("video%2F360p", 3, 3), + ] + ); + + // Audio-only playback: four adjacent groups per GET, the rest from the cache. + for segment in 0..SEGMENTS { + for (_, sequence, frames) in plan(segment).into_iter().filter(|(name, ..)| *name == "audio") { + assert_eq!(fetch(&broadcast, "audio", sequence).await.unwrap(), frames); + } + } + assert_eq!( + media(mock.gets(), "audio"), + [ + object("audio", 0, 3), + object("audio", 4, 7), + object("audio", 8, 11), + object("audio", 12, 12), + ] + ); + + // Every other enrolled group replays its original sequence, timestamps, and payloads. + for segment in 0..SEGMENTS { + for (name, sequence, frames) in plan(segment) { + assert_eq!( + fetch(&broadcast, name, sequence).await.unwrap(), + frames, + "{name} {sequence}" + ); + } + } + for (track, sequence) in [("chat", 1), ("chat", 4), ("chat", 6), ("audio", 13), ("video/1080p", 4)] { + assert!( + matches!(fetch(&broadcast, track, sequence).await, Err(moq_net::Error::NotFound)), + "{track} {sequence} was never recorded" + ); + } + assert!( + !mock.take().iter().any(|op| matches!(op, Op::List { .. })), + "a group request resolves its object without listing" + ); +} + +#[tokio::test] +async fn an_offline_reader_follows_dvr_expiry() { + let mock = Mock::memory(); + let store = Store::new(mock.clone(), "rec"); + 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 config = Config::default().with_retention(Retention::new(Duration::from_secs(2), Duration::ZERO)); + let writer = Writer::new(store.clone(), source.consume(), config).await.unwrap(); + writer.control().pacing_track("video").await.unwrap(); + let run = tokio::spawn(writer.run()); + + let frames = |sequence: u64| vec![(sequence * 1000, Bytes::from(format!("video {sequence}")))]; + let committed = |segment: u64| { + let store = store.clone(); + async move { + while store.get_segments(TIMELINE, segment).await.is_err() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + } + }; + + for sequence in 0..3 { + write(&video, sequence, &frames(sequence)); + } + committed(1).await; + + let broadcast = broadcast::Info::new().produce(); + let mut reader = Reader::open(store.clone(), &broadcast, ReaderConfig::new(TIMELINE)) + .await + .unwrap(); + tokio::spawn(reader.serve()); + assert_eq!(fetch(&broadcast, "video", 0).await.unwrap(), frames(0)); + + // The reader is offline while the DVR commits and expires several segments. + for sequence in 3..10 { + write(&video, sequence, &frames(sequence)); + } + committed(8).await; + // Segments 7 and 8 hold the 2s window. + while store.get_groups("video", 6..=6).await.is_ok() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + + mock.take(); + reader.refresh().await.unwrap(); + let ops = mock.take(); + let (list, gets) = ops.split_first().unwrap(); + assert_eq!( + list, + &Op::List { + prefix: "rec/timeline%2Ez/segments".to_string(), + offset: Some(format!("rec/timeline%2Ez/segments/{}", id(1))), + } + ); + let segments: Vec<_> = (2..=8) + .map(|segment| Op::Get(format!("rec/timeline%2Ez/segments/{}", id(segment)))) + .collect(); + assert_eq!(gets, segments, "only the new timeline keys are read"); + + // Expired groups are gone from the index, so they cost no media GET. + for sequence in 0..7 { + assert!( + matches!( + fetch(&broadcast, "video", sequence).await, + Err(moq_net::Error::NotFound) + ), + "group {sequence} expired" + ); + } + let gets = mock.gets(); + assert!(!gets.iter().any(|path| path.contains("/groups/")), "{gets:?}"); + for sequence in 7..9 { + assert_eq!(fetch(&broadcast, "video", sequence).await.unwrap(), frames(sequence)); + } + + video.finish().unwrap(); + source.finish(); + run.await.unwrap().unwrap(); +} diff --git a/rs/moq-archive/src/reader/tests.rs b/rs/moq-archive/src/reader/tests.rs index 723f02dd29..c560788199 100644 --- a/rs/moq-archive/src/reader/tests.rs +++ b/rs/moq-archive/src/reader/tests.rs @@ -331,7 +331,9 @@ async fn a_missing_tail_is_retried_on_the_next_refresh() { let mut archive = Archive::new().await; for segment in 0..3 { archive.media("video", &[(segment, 1)]).await; - archive.commit(&record(segment, &[("video", &[(segment, segment)])]), 0).await; + archive + .commit(&record(segment, &[("video", &[(segment, segment)])]), 0) + .await; } // Listed, but not yet readable. archive.store.inner().hide_gets("segments/0000000000000000002"); @@ -379,7 +381,9 @@ async fn following_lists_only_new_timeline_keys() { Err(moq_net::Error::NotFound) )); - archive.commit(&record(segment, &[("video", &[(segment, segment)])]), 0).await; + archive + .commit(&record(segment, &[("video", &[(segment, segment)])]), 0) + .await; archive.store.inner().take(); reader.refresh().await.unwrap(); let previous = format!("rec/timeline%2Ez/segments/{:019}", segment - 1); @@ -408,7 +412,9 @@ async fn an_unordered_listing_replays_in_segment_order() { for segment in 0..4 { archive.media("video", &[(segment, 1)]).await; let pop = u64::from(segment >= 2); - archive.commit(&record(segment, &[("video", &[(segment, segment)])]), pop).await; + archive + .commit(&record(segment, &[("video", &[(segment, segment)])]), pop) + .await; } let (broadcast, mut reader) = open(&archive).await; @@ -450,7 +456,10 @@ async fn a_track_without_usable_info_is_not_found() { archive.store.put_groups("bare", &object("bare")).await.unwrap(); archive.store.put_groups("future", &object("future")).await.unwrap(); archive - .raw(&Key::info("future").unwrap(), br#"{"version":2,"priority":0,"timescale":1000}"#) + .raw( + &Key::info("future").unwrap(), + br#"{"version":2,"priority":0,"timescale":1000}"#, + ) .await; archive .commit( diff --git a/rs/moq-archive/src/segment.rs b/rs/moq-archive/src/segment.rs index f2aa3fadd2..6b844dcdee 100644 --- a/rs/moq-archive/src/segment.rs +++ b/rs/moq-archive/src/segment.rs @@ -359,7 +359,11 @@ mod tests { } fn sequences(bytes: Bytes) -> Result> { - Ok(Object::decode(bytes)?.groups.iter().map(|group| group.sequence).collect()) + Ok(Object::decode(bytes)? + .groups + .iter() + .map(|group| group.sequence) + .collect()) } #[test] @@ -464,7 +468,10 @@ mod tests { assert!(matches!(table([(1, 1), (2, 1)], 3), Err(Error::Table)), "late start"); assert!(matches!(table([(0, 1), (1, 3)], 3), Err(Error::Table)), "past the end"); let varint = (1u64 << 62) - 1; - assert!(matches!(table([(0, 1), (1, varint)], 3), Err(Error::Table)), "huge length"); + assert!( + matches!(table([(0, 1), (1, varint)], 3), Err(Error::Table)), + "huge length" + ); } #[test] diff --git a/rs/moq-archive/src/store.rs b/rs/moq-archive/src/store.rs index 63cf2ec709..c79db85180 100644 --- a/rs/moq-archive/src/store.rs +++ b/rs/moq-archive/src/store.rs @@ -869,7 +869,10 @@ mod tests { ] { assert_eq!(lookup(&store, group).await, found, "group {group}"); } - store.delete(&Key::groups("video", ID_MAX..=ID_MAX).unwrap()).await.unwrap(); + store + .delete(&Key::groups("video", ID_MAX..=ID_MAX).unwrap()) + .await + .unwrap(); assert_eq!(lookup(&store, 11).await, None); assert!(Query::groups_from("video", ID_MAX + 1).is_err()); } @@ -881,11 +884,23 @@ mod tests { store.put_groups("video", &one_group(4, b"a")).await.unwrap(); assert_eq!(store.paginated_prefix(None), None); - let expected = std::collections::HashSet::from([Key::info("catalog.json").unwrap(), Key::groups("video", 4..=4).unwrap()]); - let streamed: std::collections::HashSet = store.list(&Query::new()).map_ok(|entry| entry.key).try_collect().await.unwrap(); + let expected = + std::collections::HashSet::from([Key::info("catalog.json").unwrap(), Key::groups("video", 4..=4).unwrap()]); + let streamed: std::collections::HashSet = store + .list(&Query::new()) + .map_ok(|entry| entry.key) + .try_collect() + .await + .unwrap(); assert_eq!(streamed, expected); let page = store.list_paginated(&Query::new()).await.unwrap(); - assert_eq!(page.entries.into_iter().map(|entry| entry.key).collect::>(), expected); + assert_eq!( + page.entries + .into_iter() + .map(|entry| entry.key) + .collect::>(), + expected + ); assert!(page.next.is_none()); let page = store.list_paginated(&Query::groups("video").unwrap()).await.unwrap(); assert_eq!(page.entries.len(), 1); diff --git a/rs/moq-archive/src/writer.rs b/rs/moq-archive/src/writer.rs index a46d77eac1..9ed25d86b7 100644 --- a/rs/moq-archive/src/writer.rs +++ b/rs/moq-archive/src/writer.rs @@ -705,6 +705,11 @@ impl Committer { let segment = pending.segment; let record = (*pending).clone(); + let pts = Timestamp::new( + record.pts, + Timescale::new(self.timescale).map_err(|_| Error::Timescale(self.timescale))?, + ) + .map_err(|_| Error::Id(record.pts))?; self.timeline.push(pending).map_err(timeline_error)?; self.window.push_back(record); @@ -714,7 +719,7 @@ impl Committer { } self.timeline.flush().map_err(timeline_error)?; - let object = self.read_timeline()?; + let object = self.read_timeline(pts)?; store.put_segments(&shared.timeline, segment, &object).await?; let mut keys = Vec::new(); @@ -747,8 +752,11 @@ impl Committer { expired } - /// Collect the timeline groups completed since the last segment. - fn read_timeline(&mut self) -> Result { + /// Collect the timeline groups completed since the last segment, stamped at `pts`. + /// + /// The live timeline track stamps frames with the wall clock; storing the segment's content + /// time instead keeps a recording's bytes a function of its content alone. + fn read_timeline(&mut self, pts: Timestamp) -> Result { let waiter = kio::Waiter::noop(); let timescale = self.groups.info().timescale; let mut groups = Vec::new(); @@ -759,7 +767,10 @@ impl Committer { let mut frames = Vec::new(); loop { match group.poll_read_frame(&waiter) { - Poll::Ready(Ok(Some(frame))) => frames.push(frame), + Poll::Ready(Ok(Some(mut frame))) => { + frame.timestamp = pts; + frames.push(frame); + } Poll::Ready(Ok(None)) => break, Poll::Ready(Err(err)) => return Err(timeline_error_net(err)), // Flushing closed every group, so an open one is a bug. @@ -1392,7 +1403,11 @@ mod tests { assert_eq!(ranges(&records, "video"), vec![(0, 0), (1, 1), (2, 2), (4, 4), (5, 5)]); check_objects(&store, &records).await; // An archive deletes nothing; the orphan stays invisible. - assert!(stored_groups(&store).await.contains(&Key::groups("video", 3..=3).unwrap())); + assert!( + stored_groups(&store) + .await + .contains(&Key::groups("video", 3..=3).unwrap()) + ); assert!(!referenced(&records).contains(&Key::groups("video", 3..=3).unwrap())); } From 9005ed63cf4e91004c57866c85aeee4edbf00b84 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:47:35 -0700 Subject: [PATCH 07/10] fix(archive): cache each served track's .info Since #4104 an idle track is re-requested, so a repeated segment request GET .info again and broke the HLS archive cache test. The object is immutable, so keep the parsed info once it loads. Co-Authored-By: Claude Opus 5.5 --- rs/moq-archive/src/reader/mod.rs | 25 ++++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/rs/moq-archive/src/reader/mod.rs b/rs/moq-archive/src/reader/mod.rs index 64b22c9fab..5be88a433c 100644 --- a/rs/moq-archive/src/reader/mod.rs +++ b/rs/moq-archive/src/reader/mod.rs @@ -23,6 +23,7 @@ mod index; +use std::collections::HashMap; use std::future::Future; use std::ops::RangeInclusive; use std::sync::{Arc, Mutex}; @@ -79,6 +80,8 @@ struct Shared { store: Store, index: Mutex, cache: quick_cache::sync::Cache, Weight>, + /// Each track's `.info`, which is immutable, so a re-requested track costs no GET. + infos: Mutex>, } impl Reader { @@ -97,6 +100,7 @@ impl Reader { store, index: Mutex::default(), cache: quick_cache::sync::Cache::with_weighter(items, config.cache, Weight), + infos: Mutex::default(), }); let mut reader = Self { @@ -237,13 +241,20 @@ async fn serve_track(shared: Arc>, request: track::Req return; } - let info = match shared.store.get_info(&name).await.and_then(|info| track_info(&info)) { - Ok(info) => info, - Err(err) => { - tracing::warn!(track = name, %err, "archived track has no usable .info"); - request.reject(moq_net::Error::NotFound); - return; - } + let cached = shared.infos.lock().unwrap().get(&name).cloned(); + let info = match cached { + Some(info) => info, + None => match shared.store.get_info(&name).await.and_then(|info| track_info(&info)) { + Ok(info) => { + shared.infos.lock().unwrap().insert(name.clone(), info.clone()); + info + } + Err(err) => { + tracing::warn!(track = name, %err, "archived track has no usable .info"); + request.reject(moq_net::Error::NotFound); + return; + } + }, }; let timescale = info.timescale; From 266077383764c9c3605e497945c13c0eca7cd68f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:47:35 -0700 Subject: [PATCH 08/10] test(hls): refuse media GETs while rendering archive playlists Playlists now render against a store that fails every media GET, and segment requests are pinned to issue no listing. Co-Authored-By: Claude Opus 5.5 --- rs/moq-hls/src/export/archive_tests.rs | 35 ++++++++++++++++++++++---- 1 file changed, 30 insertions(+), 5 deletions(-) diff --git a/rs/moq-hls/src/export/archive_tests.rs b/rs/moq-hls/src/export/archive_tests.rs index 35e05e2592..40f327c8f6 100644 --- a/rs/moq-hls/src/export/archive_tests.rs +++ b/rs/moq-hls/src/export/archive_tests.rs @@ -5,6 +5,7 @@ //! pin the storage traffic that composition produces: playlists read only the timeline, and a //! segment GETs exactly one object of the requested rendition. +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -24,11 +25,14 @@ use super::*; const TIMELINE: &str = hang::timeline::DEFAULT_NAME; -/// In-memory store that records every GET path. +/// In-memory store that records every GET path and counts listings. #[derive(Debug, Clone, Default)] struct Counting { inner: Arc, gets: Arc>>, + lists: Arc, + /// Fail every media GET, so nothing can quietly depend on one. + reject_media: Arc, } impl Counting { @@ -36,6 +40,15 @@ impl Counting { fn take(&self) -> Vec { std::mem::take(&mut *self.gets.lock().unwrap()) } + + /// Listings since the store was created. + fn lists(&self) -> usize { + self.lists.load(Ordering::SeqCst) + } + + fn reject_media(&self, reject: bool) { + self.reject_media.store(reject, Ordering::SeqCst); + } } impl std::fmt::Display for Counting { @@ -65,6 +78,12 @@ impl ObjectStore for Counting { async fn get_opts(&self, location: &Path, options: GetOptions) -> object_store::Result { self.gets.lock().unwrap().push(location.to_string()); + if self.reject_media.load(Ordering::SeqCst) && is_media(location.as_ref()) { + return Err(object_store::Error::NotImplemented { + operation: "media GET".into(), + implementer: "Counting".into(), + }); + } self.inner.get_opts(location, options).await } @@ -76,6 +95,7 @@ impl ObjectStore for Counting { } fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result> { + self.lists.fetch_add(1, Ordering::SeqCst); self.inner.list(prefix) } @@ -342,7 +362,9 @@ async fn playlists_read_only_the_timeline_and_segments_one_object() { let master = replay.broadcaster.master_playlist(None); assert!(master.contains("video/360p/media.m3u8") && master.contains("video/1080p/media.m3u8")); - // Render and reload every playlist: aligned numbering, and not one media GET. + // Render and reload every playlist with media GETs refused: aligned numbering, and not one + // media GET. + recording.store.inner().reject_media(true); for _ in 0..2 { for (kind, name) in [(Kind::Video, "360p"), (Kind::Video, "1080p"), (Kind::Audio, "audio")] { let playlist = replay.playlist(kind, name).await; @@ -358,6 +380,9 @@ async fn playlists_read_only_the_timeline_and_segments_one_object() { !gets.iter().any(|path| is_media(path)), "playlists must not GET media: {gets:?}" ); + recording.store.inner().reject_media(false); + // A range-bearing segment URI resolves its object directly: no listing, no index object. + let lists = recording.store.inner().lists(); // Switching renditions downloads only the selected rendition's object. let low = replay.rendition(Kind::Video, "360p").segment(1).await.unwrap().unwrap(); @@ -400,10 +425,10 @@ async fn playlists_read_only_the_timeline_and_segments_one_object() { ] ); - // A repeated request hits the reader's object cache. The parked track re-subscribes - // upstream to confirm its warm cache, which re-reads only the small `.info`. + // A repeated request hits the reader's cache, including the immutable `.info`. replay.rendition(Kind::Video, "360p").segment(1).await.unwrap().unwrap(); - assert_eq!(recording.gets(), ["rec/360p/.info"]); + assert_eq!(recording.gets(), Vec::::new()); + assert_eq!(recording.store.inner().lists(), lists, "segments never list"); } #[tokio::test] From 89c17f591f83e2e793aa7cca703ba6854fe30a96 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:48:26 -0700 Subject: [PATCH 09/10] quest(archive): complete the proof Co-Authored-By: Claude Opus 5.5 --- quest/m1/archive/README.md | 4 ++- quest/m1/archive/proof.md | 67 -------------------------------------- 2 files changed, 3 insertions(+), 68 deletions(-) delete mode 100644 quest/m1/archive/proof.md diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index 89cd0aedad..02157e86ca 100644 --- a/quest/m1/archive/README.md +++ b/quest/m1/archive/README.md @@ -60,6 +60,9 @@ 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. +`rs/moq-archive/src/proof.rs` records one multi-rendition broadcast end to end: its exact keys and +bytes match on memory, local disk, and an unordered listing, FETCH replays every group exactly, +and a rendition's playback GETs only that rendition's objects. ### Format @@ -133,7 +136,6 @@ owned by that prerequisite, not duplicated in archive storage. - [Replay catalog](/quest/m1/archive/replay-catalog.md) - `moq import archive` publishes the recorded catalog live with `store` set, so stock `moq export hls` serves the whole replay - [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/proof.md b/quest/m1/archive/proof.md deleted file mode 100644 index b24575dbb1..0000000000 --- a/quest/m1/archive/proof.md +++ /dev/null @@ -1,67 +0,0 @@ -# [M] Archive proof - -## Goal - -Prove deterministic segment storage, exact FETCH replay, selective rendition -reads, and timeline-only HLS generation from one multi-rendition broadcast. - -## Plan - -Record explicitly enrolled video, audio, catalog, and non-media tracks. Cut each -track on its own timeline, including many audio groups per object and a group -split across objects by frame, then -replay their original sequences, timestamps, and payloads through -`track::Dynamic`. - -Verify the exact object keys and bytes on memory, local, and S3-compatible -`object_store` implementations, including the percent-encoded track names. A -360p or audio-only FETCH must not GET the 1080p object, while adjacent group -requests should hit the segment LRU. - -Cover the persistence boundary: a crash after segment PUT but before timeline -commit leaves invisible orphan data that a listing bootstrap ignores; a failed -or mismatched `.info` exposes no ranges; a failed independent track PUT omits -only that track while the record's other tracks stay advertised; later -segments remain usable. Catalog-to-group applicability is outside this proof. Also -cover a stalled pacing track forced to a gap, sparse group ranges, malformed -table offsets, an unknown envelope or `.info` version treated as a missing -segment, segment create collisions under the single-writer rule, a missing -tail, and a clean end without a completion marker. Accept equivalent `.info` -JSON with reordered members or different whitespace without rewriting it; -reject differing properties and malformed or unsupported metadata. - -Exercise group and segment IDs 0 and 2^53 - 1; reject 2^53 and the largest -QUIC varint in keys and reconstructed group IDs. Cover consecutive zero -deltas, sparse deltas, overflow, stopping at ID exhaustion without wrapping -or inferring clean finality, empty objects, mismatched filename bounds, and overlapping -ranges across segments. Reject decreasing or duplicate group arrivals while -allowing accepted groups to complete out of order. Verify JSON-safe timescales -and timestamps, accepting timestamp 2^53 - 1 and rejecting 2^53. -Test ordered S3 lookup at both endpoints and between ranges, unordered paginated -results, incremental cursor recovery, DVR expiration while a reader is offline, -and stale media listings preceding a new timeline commit. Following N+1 must -not refresh all media listings. Wire these cases into CI for the store, writer, -and reader implementations; do not add an unconnected standalone proof script. - -The store's own tests also cover recording-prefix isolation (`rec` beside -`rec-other`), empty prefixes, track-prefix listings, continuation pages, and -every supported pagination option. A page must not lose directory entries -silently or fail because the backend matched a neighbouring recording. Every -publicly constructible key either serializes to a path its parser accepts or -fails before storage; direct `Key::Groups` construction must not bypass range -validation. These cases belong beside the store and codec code and need no -new public API or format change. - -Crash a DVR writer after its pop becomes durable but before media deletion. -On exclusive restart, prove that expired and uncommitted group objects are -removed after the grace period while retained media, `.info`, and checkpoint -objects survive. Failed or incomplete recovery/listing must delete nothing; -restart must finish this cleanup before accepting new groups. - -Finally render and reload HLS playlists while rejecting every media-object GET -until a segment URI is requested. A segment request must resolve one object -from the replayed timeline without any listing or separate index object. - -## Required - -- [Rust per-track timelines](/quest/m1/archive/track-timeline/core.md) - proves the per-track format, not the aligned one From 975f97b3082abf3126caa4f192e0044de16d6280 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:49:19 -0700 Subject: [PATCH 10/10] quest(archive): follow up with DVR timeline pruning and an S3 wire proof Co-Authored-By: Claude Opus 5.5 --- quest/m1/archive/README.md | 1 + quest/m1/archive/pruning.md | 17 +++++++++++++++++ quest/m2/README.md | 1 + quest/m2/archive-s3.md | 22 ++++++++++++++++++++++ 4 files changed, 41 insertions(+) create mode 100644 quest/m1/archive/pruning.md create mode 100644 quest/m2/archive-s3.md diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index 02157e86ca..1f3a9ff4d7 100644 --- a/quest/m1/archive/README.md +++ b/quest/m1/archive/README.md @@ -136,6 +136,7 @@ owned by that prerequisite, not duplicated in archive storage. - [Replay catalog](/quest/m1/archive/replay-catalog.md) - `moq import archive` publishes the recorded catalog live with `store` set, so stock `moq export hls` serves the whole replay - [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 +- [DVR timeline pruning](/quest/m1/archive/pruning.md) - a DVR deletes timeline objects no retained checkpoint needs ## Related diff --git a/quest/m1/archive/pruning.md b/quest/m1/archive/pruning.md new file mode 100644 index 0000000000..3e2dfcfbc6 --- /dev/null +++ b/quest/m1/archive/pruning.md @@ -0,0 +1,17 @@ +# [S] DVR timeline pruning + +## Goal + +A DVR writer deletes timeline objects that no retained checkpoint needs, so a +long-running DVR stores a bounded number of objects. + +## Plan + +`moq_archive::Writer` deletes expired group objects but never a +`segments/` timeline object, so a 24/7 DVR with 2s segments adds +about 43,000 objects a day, and every restart lists all of them. The draft +already allows it: the writer keeps the latest timeline object and enough +earlier groups to recover the retained window from a checkpoint. Delete the +oldest timeline objects no longer needed, one grace period after they stop +being needed, keeping the remaining keys contiguous, as recovery requires. +Extend the restart cleanup and its tests in `rs/moq-archive/src/writer.rs`. diff --git a/quest/m2/README.md b/quest/m2/README.md index f7cc1a225c..4cd3ed3c47 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -18,6 +18,7 @@ upstream release waits in [m4](/quest/m4/README.md). - [AV1 metadata separation](/quest/m2/av1-metadata.md) - retain metadata OBUs inline while evaluating separate delivery - [SEI separation](/quest/m2/sei/README.md) - retain inline SEI until measured savings or a metadata-only consumer justify a split - [Catalog track identity](/quest/m2/catalog-tracks.md) - compare immutable track definitions with explicit catalog-to-group binding +- [Archive S3 wire proof](/quest/m2/archive-s3.md) - the archive proof also runs through the S3 client against an in-process S3-compatible server - [Mobile ownership](/quest/m2/mobile-ownership.md) - decide whether Rust or platform code owns mobile capture, codecs, and rendering - [iOS capture](/quest/m2/mobile-capture-ios.md) - camera and screen capture if the mobile ownership decision selects Rust - [Android capture](/quest/m2/mobile-capture-android.md) - NDK/JNI capture using the existing codecs if mobile ownership selects Rust diff --git a/quest/m2/archive-s3.md b/quest/m2/archive-s3.md new file mode 100644 index 0000000000..d190490d8c --- /dev/null +++ b/quest/m2/archive-s3.md @@ -0,0 +1,22 @@ +# [S] Archive S3 wire proof + +## Goal + +The `moq-archive` proof (`rs/moq-archive/src/proof.rs`) also runs through +`object_store`'s S3 client against an in-process S3-compatible server, in CI +with no external network. + +## Plan + +The proof covers memory, local disk, and an S3-style listing fake +(`rs/moq-archive/src/mock.rs`), but not the S3 client: percent-encoded keys +such as `catalog%2Ejson` inside request URLs, `PutMode::Create` through +`If-None-Match`, `start-after` listing, and continuation tokens. Evaluate a +maintained in-process server (for example `s3s` with `s3s-fs`) on loopback; +refuse one that ignores conditional creates, since collisions would pass +silently. Gate it behind a dev-dependency feature if the build cost is large, +and wire it into at least the nightly workflow. + +## Related + +- [Timeline-indexed MoQ archives](/quest/m1/archive/README.md)