diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index 89cd0aedad..1f3a9ff4d7 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,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 -- [Archive proof](/quest/m1/archive/proof.md) - prove persistence ordering, selective reads, exact FETCH replay, and timeline-only HLS generation +- [DVR timeline pruning](/quest/m1/archive/pruning.md) - a DVR deletes timeline objects no retained checkpoint needs ## 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 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) diff --git a/rs/moq-archive/src/lib.rs b/rs/moq-archive/src/lib.rs index 3d5d9bb1d8..3edecbe87e 100644 --- a/rs/moq-archive/src/lib.rs +++ b/rs/moq-archive/src/lib.rs @@ -17,7 +17,11 @@ pub use object_store; mod error; 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 new file mode 100644 index 0000000000..2d7275badf --- /dev/null +++ b/rs/moq-archive/src/mock.rs @@ -0,0 +1,272 @@ +//! 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/path.rs b/rs/moq-archive/src/path.rs index 86a8294e7b..4df4784b40 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,52 @@ 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/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/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; diff --git a/rs/moq-archive/src/reader/tests.rs b/rs/moq-archive/src/reader/tests.rs index ca817bf72a..c560788199 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, Op}; 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] @@ -396,6 +326,161 @@ 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; @@ -440,7 +525,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/segment.rs b/rs/moq-archive/src/segment.rs index c2fd68c5aa..6b844dcdee 100644 --- a/rs/moq-archive/src/segment.rs +++ b/rs/moq-archive/src/segment.rs @@ -346,6 +346,39 @@ 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 +446,34 @@ 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 3a9092703e..c79db85180 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") } @@ -520,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(); @@ -818,7 +745,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 +754,6 @@ mod tests { } store .inner() - .inner .put( &Path::from("rec-other/video/.info"), Bytes::from_static(b"sibling").into(), @@ -896,6 +822,90 @@ 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(); diff --git a/rs/moq-archive/src/writer.rs b/rs/moq-archive/src/writer.rs index 5f05cdc083..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. @@ -803,91 +814,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 +963,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(); @@ -1125,6 +1054,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(); @@ -1365,20 +1327,101 @@ 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 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(_)))); 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]