From ba2ae8e519c0fcbac10e65865f8d56b31ce2440e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 22:50:39 -0700 Subject: [PATCH 1/3] quest: claim quest/m1/archive/cli From 47f59984fec2a9569b4a05f915ec03058d7d8c4f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 23:23:45 -0700 Subject: [PATCH 2/3] feat(cli): record and replay archives with `export archive` / `import archive` `export archive ` records one broadcast through moq_archive::Writer, enrolling catalog renditions as they appear. `import archive ` republishes a recording through moq_archive::Reader, optionally following a growing archive with `--follow`. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 39 ++++ Cargo.toml | 1 + doc/bin/cli.md | 35 +++- quest/m1/archive/README.md | 1 - quest/m1/archive/cli.md | 43 ---- rs/moq-cli/Cargo.toml | 15 +- rs/moq-cli/README.md | 2 +- rs/moq-cli/src/archive.rs | 383 ++++++++++++++++++++++++++++++++++++ rs/moq-cli/src/args.rs | 6 +- rs/moq-cli/src/main.rs | 18 ++ rs/moq-cli/src/subscribe.rs | 8 + 11 files changed, 499 insertions(+), 52 deletions(-) delete mode 100644 quest/m1/archive/cli.md create mode 100644 rs/moq-cli/src/archive.rs diff --git a/Cargo.lock b/Cargo.lock index 4754c84d2b..0c1fc204ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1471,6 +1471,16 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" +[[package]] +name = "crc-fast" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e75b2483e97a5a7da73ac68a05b629f9c53cff58d8ed1c77866079e18b00dba5" +dependencies = [ + "digest 0.10.7", + "spin 0.10.1", +] + [[package]] name = "crc32fast" version = "1.5.2" @@ -4050,6 +4060,16 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest 0.11.3", +] + [[package]] name = "mdns-sd" version = "0.21.4" @@ -4298,6 +4318,7 @@ dependencies = [ "bytes", "hang", "humantime", + "moq-archive", "moq-audio", "moq-auth", "moq-hls", @@ -4310,6 +4331,7 @@ dependencies = [ "moq-tokio", "moq-transcode", "moq-video", + "object_store", "pollster", "reqwest", "rustls", @@ -6174,17 +6196,32 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1796bc93603f78c5760a69f2d58badc9618d22adade0a95385bb2adbae4eb94" dependencies = [ "async-trait", + "aws-lc-rs", + "base64 0.23.1", "bytes", "chrono", + "crc-fast", + "form_urlencoded", "futures-channel", "futures-core", "futures-util", "http", + "http-body-util", + "httparse", "humantime", + "hyper", "itertools 0.15.0", + "md-5", "nix", "parking_lot", "percent-encoding", + "quick-xml 0.41.0", + "rand 0.10.3", + "reqwest", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", "thiserror 2.0.21", "tokio", "tracing", @@ -6875,6 +6912,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" dependencies = [ "memchr", + "serde", ] [[package]] @@ -7218,6 +7256,7 @@ dependencies = [ "bytes", "futures-core", "futures-util", + "h2", "http", "http-body", "http-body-util", diff --git a/Cargo.toml b/Cargo.toml index 3e25cae5ee..f6b12a60e4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -140,6 +140,7 @@ loom = { version = "0.7.2", features = ["futures"] } # `async` awaits the event channel instead of blocking a thread on it. mdns-sd = { version = "0.21", features = ["async"] } moq-audio = { version = "0.1.2", path = "rs/moq-audio", default-features = false } +moq-archive = { version = "0.0.3", path = "rs/moq-archive" } moq-auth = { version = "0.1.1", path = "rs/moq-auth" } moq-binary = { version = "0.1.2", path = "rs/moq-binary" } moq-flate = { version = "0.2.0", path = "rs/moq-flate" } diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 1376686f9d..7c8ecde6ca 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -18,9 +18,11 @@ or Docker; see [Install](/setup/install). | `import` | `capture` | Capture a camera, display, window, or app plus a microphone, and encode natively. | | `import` | `hls ` | Pull a remote HLS playlist. | | `import` | `rtmp`, `srt`, `rtc` | Accept pushes (`--listen`) or pull from a remote (`--connect`). | +| `import` | `archive ` | Replay a recording from an object store. | | `export` | `fmp4`, `mkv`, `ts`, `flv`, `h264`, `h265` | Write a container to stdout. | | `export` | `hls --listen` | Serve the broadcast as HLS over HTTP. | | `export` | `rtmp`, `srt`, `rtc` | Serve plays (`--listen`) or push to a remote (`--connect`). | +| `export` | `archive ` | Record the broadcast into an object store. | | `play` | | Decode and play in a native window with sound. | | `transcode` | | Publish a just-in-time rendition ladder next to a broadcast. | | `fetch` | `` | Write one group of a track to stdout. | @@ -178,6 +180,36 @@ refuses a listener or cluster flag. It gives up after 30 seconds, as `/fetch` does, and exits non-zero when the broadcast or group is not found, the relay refuses, or the deadline passes. +## Archive + +```bash +# Record a broadcast until it ends +moq --connect https://relay.example.com/anon --broadcast event.hang export archive s3://recordings/event + +# Replay it under another name +moq --connect https://relay.example.com/anon --broadcast event-replay.hang import archive s3://recordings/event +``` + +`export archive` records one broadcast with +[moq-archive](https://docs.rs/moq-archive), reading its catalog as it changes: +video and audio renditions pace the segments, and the catalog plus every text, +JSON, and binary track are recorded alongside. It refuses a rendition served +from another broadcast, and one that returns after the catalog dropped it. The +stage ends once the broadcast does, and it refuses a store URL that already +holds a recording. `--retention 1h` keeps only the last hour (a DVR), +deleting expired objects `--retention-grace` (default 30s) after the timeline +stops advertising them. + +`import archive` republishes a recording: the timeline replays as a live track +and every other track's groups are served on request, one object GET per group +range. By default it replays what is stored and ends the timeline there; +`--follow 2s` keeps checking for new segments of a recording still being made. + +Store URLs are `file:///absolute/path`, `s3://bucket/prefix`, +`gs://bucket/prefix`, or `az://container/prefix`. Cloud credentials come from +the usual `AWS_*`, `GOOGLE_*`, and `AZURE_*` environment variables. The `s3`, +`gcs`, and `azure` cargo features are on by default. + ## Multiple stages Separate stages with `--` to bridge several broadcasts, or both directions, @@ -186,7 +218,8 @@ over one connection: ```bash moq --connect https://relay.example.com/anon \ import --broadcast event.hang srt --listen 0.0.0.0:9000 \ - -- export --broadcast event.hang hls --listen 0.0.0.0:8080 + -- export --broadcast event.hang hls --listen 0.0.0.0:8080 \ + -- export --broadcast event.hang archive file:///recordings/event ``` ## Redundant publishers diff --git a/quest/m1/archive/README.md b/quest/m1/archive/README.md index 80e59ac56c..e4cdd70249 100644 --- a/quest/m1/archive/README.md +++ b/quest/m1/archive/README.md @@ -111,7 +111,6 @@ owned by that prerequisite, not duplicated in archive storage. ## Quests -- [Archive endpoint](/quest/m1/archive/cli.md) - `moq ... export archive` records and `import archive` replays - [Browser archive](/quest/m1/archive/browser.md) - the same contract for browser-published broadcasts - [Offline archive HLS](/quest/m1/archive/hls.md) - render playlists from the archive timeline and fetch segment media lazily - [Resume a recording](/quest/m1/archive/recovery.md) - recover the retained timeline on restart and clean up DVR orphans diff --git a/quest/m1/archive/cli.md b/quest/m1/archive/cli.md deleted file mode 100644 index cfa76925bf..0000000000 --- a/quest/m1/archive/cli.md +++ /dev/null @@ -1,43 +0,0 @@ -# [M] Archive endpoint in moq-cli - -## Goal - -`moq-cli` records and replays archives with no custom code, through an -`archive ` endpoint on the existing direction verbs: -`export archive` records a broadcast with `moq_archive::Writer`, and -`import archive` republishes a recording, serving its groups on demand -through `moq_archive::Reader`. Recording an ingest in one process is a -multi-stage line. - -```bash -moq --connect $RELAY --broadcast ev.hang export archive s3://rec/ev -moq --connect $RELAY \ - import --broadcast ev.hang srt --listen 0.0.0.0:9000 \ - -- export --broadcast ev.hang archive file:///rec/ev -moq --connect $RELAY --broadcast ev-replay.hang import archive s3://rec/ev -``` - -## Plan - -Parse the store URL with `object_store::parse_url`. Compile in every -`object_store` cloud backend (S3, GCS, Azure) behind features, with local -files always available. - -An archive stage carries exactly one broadcast; refuse a multi-broadcast -stage (an RTMP or SRT listener accepting many) rather than invent a prefix -scheme. - -Export reads its own catalog: video and audio renditions enroll as pacing -tracks, the catalog and any other track as non-pacing, and renditions added -later enroll as they appear. The writer stays catalog-agnostic. Expose the -writer's retention window and deletion grace as flags. - -Import hands the reader the stage's `broadcast::Producer`. The recording is -not live; decide with the reader whether import finishes the replayed -timeline or keeps following a growing archive. - -Update `doc/bin/cli.md` and every example invocation. - -## Closes - -- [#2281](https://github.com/moq-dev/moq/issues/2281) - close this issue when the quest finishes diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index de9b25aa66..f5830ff03e 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -26,7 +26,7 @@ path = "src/main.rs" doc = false [features] -default = ["iroh", "cluster-lan", "noq", "websocket", "nvidia"] +default = ["iroh", "cluster-lan", "noq", "websocket", "nvidia", "s3", "gcs", "azure"] iroh = ["moq-tokio/iroh"] # LAN discovery and meshing (`--cluster-lan`). cluster-lan = ["moq-relay/cluster-lan"] @@ -37,6 +37,11 @@ noq = ["moq-tokio/noq"] # congestion-control testing. qlog = ["moq-tokio/qlog"] websocket = ["moq-tokio/websocket"] +# The cloud object stores `export archive` and `import archive` reach (`s3://`, +# `gs://`, `az://`). Local `file://` URLs need none of them. +s3 = ["object_store/aws"] +gcs = ["object_store/gcp"] +azure = ["object_store/azure"] # Device capture (camera + microphone) + encode/publish. Off by default because # it pulls in moq-video + moq-audio capture: on Linux the microphone needs # ALSA/libasound at build time through cpal (the camera needs nothing, moq-v4l @@ -93,10 +98,13 @@ base64 = { workspace = true } bytes = { workspace = true } hang = { workspace = true } humantime = { workspace = true } +moq-archive = { workspace = true } moq-audio = { workspace = true, optional = true, features = ["aac"] } moq-auth = { workspace = true, features = ["serve"] } # `server` enables the HTTP egress server for `moq export hls`; the importer is always available. moq-hls = { workspace = true, features = ["server"] } +# Names the MSF catalog track `export archive` records. +moq-msf = { workspace = true } moq-mux = { workspace = true } # Cluster, LAN mesh, and inbound `/.cluster` auth. default-features off so the # CLI's own feature set (iroh, noq, cluster-lan) is what selects them. @@ -108,6 +116,7 @@ moq-srt = { workspace = true } moq-tokio = { workspace = true, default-features = false, features = ["aws-lc-rs", "tcp", "uds"] } moq-transcode = { workspace = true, optional = true, features = ["openh264"] } moq-video = { workspace = true, optional = true, features = ["openh264"] } +object_store = { workspace = true, features = ["fs"] } pollster = { workspace = true, optional = true } reqwest = { workspace = true, features = ["rustls", "json"] } rustls = { version = "0.23", features = ["aws-lc-rs"], default-features = false } @@ -124,10 +133,6 @@ winit = { version = "0.30.13", optional = true } sd-notify = { workspace = true } [dev-dependencies] -# Authors an MSF-only catalog, which is the one shape that tells a completer reading -# the wrong catalog track apart from one reading the right one: `moq-mux`'s producer -# publishes hang and MSF from the same source, so both answer for an ordinary broadcast. -moq-msf = { path = "../moq-msf" } # `test_relay` stands up a real relay for the `fetch` tests to read through. moq-relay = { path = "../moq-relay", default-features = false, features = ["test-support"] } tempfile = { workspace = true } diff --git a/rs/moq-cli/README.md b/rs/moq-cli/README.md index de0241a778..95fd4ef6a7 100644 --- a/rs/moq-cli/README.md +++ b/rs/moq-cli/README.md @@ -19,7 +19,7 @@ Multi-arch images (`linux/amd64` and `linux/arm64`) are published to [Docker Hub ## Usage -`moq-cli` routes endpoints onto a shared MoQ Origin: `moq `. The MoQ side (before the verb) dials with `--connect `, self-hosts QUIC/WebTransport with `--listen `, or self-hosts raw qmux with `--listen-tcp-bind ` / `--listen-unix-bind ` (Unix only). `import` moves media into MoQ, `export` moves it out. The endpoint is a container format (`fmp4`, `ts`, `flv`, ... read from stdin / written to stdout), or a gateway (`hls`, `rtmp`, `srt`, `rtc`). A build with the `play` feature can also render a broadcast locally with `moq play`. +`moq-cli` routes endpoints onto a shared MoQ Origin: `moq `. The MoQ side (before the verb) dials with `--connect `, self-hosts QUIC/WebTransport with `--listen `, or self-hosts raw qmux with `--listen-tcp-bind ` / `--listen-unix-bind ` (Unix only). `import` moves media into MoQ, `export` moves it out. The endpoint is a container format (`fmp4`, `ts`, `flv`, ... read from stdin / written to stdout), or a gateway (`hls`, `rtmp`, `srt`, `rtc`, `archive`). A build with the `play` feature can also render a broadcast locally with `moq play`. Separate additional stages with `--` to bridge several broadcasts (or both directions) over one connection, each naming its own `--broadcast`: diff --git a/rs/moq-cli/src/archive.rs b/rs/moq-cli/src/archive.rs new file mode 100644 index 0000000000..f8eec6949c --- /dev/null +++ b/rs/moq-cli/src/archive.rs @@ -0,0 +1,383 @@ +//! Archive endpoints: record one broadcast into an object store (export), or +//! republish a recording from one (import), through `moq-archive`. + +use std::collections::HashSet; +use std::time::Duration; + +use anyhow::Context; +use hang::moq_net; +use moq_archive::writer::{Control, Retention}; +use moq_mux::catalog::{CatalogFormat, Stream as _}; +use object_store::ObjectStore; +use url::Url; + +use crate::moq::notify_ready; + +/// How long expired objects outlive the timeline that dropped them, unless +/// `--retention-grace` says otherwise. +const GRACE: Duration = Duration::from_secs(30); + +/// `export archive` args. +#[derive(usage::Args, Clone)] +#[usage(unknown_flags = "error", args_override_self = false)] +pub struct ExportArgs { + /// Where to record: `file:///dir`, `s3://bucket/prefix`, `gs://bucket/prefix`, or `az://container/prefix`. + pub store: Url, + + /// Keep only this much recent content (a DVR), deleting older segments. Unset keeps everything. + #[usage(long)] + pub retention: Option, + + /// How long expired objects outlive the timeline that dropped them, so readers can finish + /// their downloads. Defaults to 30s; needs `--retention`. + #[usage(long)] + pub retention_grace: Option, +} + +/// `import archive` args. +#[derive(usage::Args, Clone)] +#[usage(unknown_flags = "error", args_override_self = false)] +pub struct ImportArgs { + /// The recording to replay, as `export archive` wrote it. + pub store: Url, + + /// Keep following a recording that is still growing, checking for new segments at this + /// interval. Unset replays what is stored now and ends the timeline there. + #[usage(long)] + pub follow: Option, +} + +/// Record the broadcast `name` into `args.store` until the broadcast ends. +/// +/// Reads the broadcast's own catalog: video and audio renditions pace the segments, while the +/// catalog and every other track are recorded alongside without pacing. +pub async fn export( + origin: moq_net::origin::Consumer, + name: String, + format: CatalogFormat, + args: ExportArgs, +) -> anyhow::Result<()> { + let retention = match (args.retention, args.retention_grace) { + (Some(window), grace) => Some(Retention::new( + window.into_std(), + grace.map_or(GRACE, crate::duration::Duration::into_std), + )), + (None, Some(_)) => anyhow::bail!("`--retention-grace` needs `--retention`"), + (None, None) => None, + }; + let catalog_track = catalog_track(format)?; + let store = open(&args.store)?; + + let broadcast = origin + .routed_broadcast(&name) + .await + .with_context(|| format!("broadcast `{name}` is unavailable"))?; + let config = moq_archive::writer::Config::default().with_retention(retention); + let writer = moq_archive::Writer::new(store, broadcast.clone(), config) + .await + .with_context(|| format!("failed to start recording into {}", args.store))?; + let control = writer.control(); + control.track(catalog_track).await?; + let catalog = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await?; + + tracing::info!(%name, store = %args.store, "recording"); + notify_ready(); + + // The writer decides when the recording is over; the catalog only feeds it tracks. + tokio::select! { + result = writer.run() => result.context("recording failed"), + Err(err) = enroll(control, catalog) => Err(err), + } +} + +/// Republish the recording at `args.store` as the broadcast `name`. +/// +/// The timeline replays as a live track and every other track's groups are served on request. +pub async fn import(origin: moq_net::origin::Producer, name: String, args: ImportArgs) -> anyhow::Result<()> { + let store = open(&args.store)?; + let broadcast = origin.create_broadcast(&name).context("failed to create broadcast")?; + let config = moq_archive::reader::Config::new(hang::timeline::DEFAULT_NAME); + let mut reader = moq_archive::Reader::open(store, &broadcast, config) + .await + .with_context(|| format!("no readable recording at {}", args.store))?; + let serve = reader.serve(); + broadcast + .announce(Default::default()) + .context("failed to announce broadcast")?; + + tracing::info!(%name, store = %args.store, "replaying"); + notify_ready(); + + // The store holds no end marker, so only the caller can say the recording is complete. + let Some(interval) = args.follow.map(crate::duration::Duration::into_std) else { + reader.finish()?; + serve.await; + return Ok(()); + }; + let refresh = async { + loop { + tokio::time::sleep(interval).await; + reader.refresh().await?; + } + }; + tokio::select! { + () = serve => Ok(()), + result = refresh => result, + } +} + +/// Open the object store `url` names, taking cloud credentials from the `AWS_*`, `GOOGLE_*`, +/// and `AZURE_*` environment variables. +fn open(url: &Url) -> anyhow::Result>> { + let credentials = std::env::vars_os() + .filter_map(|(key, value)| Some((key.into_string().ok()?, value.into_string().ok()?))) + .filter(|(key, _)| ["AWS_", "GOOGLE_", "AZURE_"].iter().any(|cloud| key.starts_with(cloud))); + let (store, prefix) = + object_store::parse_url_opts(url, credentials).with_context(|| format!("unsupported archive store {url}"))?; + Ok(moq_archive::Store::new(store, prefix)) +} + +/// The track carrying the catalog `format` reads. +fn catalog_track(format: CatalogFormat) -> anyhow::Result<&'static str> { + Ok(match format { + CatalogFormat::Hang => hang::Catalog::DEFAULT_NAME, + CatalogFormat::HangZ => hang::Catalog::COMPRESSED_NAME, + CatalogFormat::Msf => moq_msf::DEFAULT_NAME, + other => anyhow::bail!("`export archive` cannot record a {other:?} catalog"), + }) +} + +/// Enroll each track as the catalog lists it, until the catalog ends. +async fn enroll( + control: Control, + mut catalog: moq_mux::catalog::Consumer<()>, +) -> anyhow::Result<()> { + let mut tracks = Tracks::default(); + while let Some(snapshot) = catalog.next().await? { + for change in tracks.update(&snapshot)? { + match change { + Change::Pacing(name) => control.pacing_track(&name).await?, + Change::Track(name) => control.track(&name).await?, + Change::Remove(name) => control.remove(&name)?, + } + } + } + Ok(()) +} + +/// A change to the recorded track set. +#[derive(Debug, PartialEq, Eq)] +enum Change { + /// Record a rendition that paces the segments. + Pacing(String), + /// Record a track without letting it pace the segments. + Track(String), + /// Stop recording a track the catalog dropped. + Remove(String), +} + +/// The catalog tracks being recorded, diffed against each new snapshot. +#[derive(Default)] +struct Tracks { + /// Every name ever enrolled. The writer never takes a name back. + enrolled: HashSet, + /// The names the latest snapshot listed. + listed: HashSet, +} + +impl Tracks { + /// The changes that bring the recording in line with `catalog`. + /// + /// Refuses a track served from another broadcast, since an archive holds exactly one, and a + /// track that returns after the catalog dropped it, since the writer cannot resume one. + fn update(&mut self, catalog: &hang::Catalog) -> anyhow::Result> { + let video = catalog + .video + .renditions + .iter() + .map(|(name, c)| (name, &c.broadcast, true)); + let audio = catalog + .audio + .renditions + .iter() + .map(|(name, c)| (name, &c.broadcast, true)); + let text = catalog + .text + .renditions + .iter() + .map(|(name, c)| (name, &c.broadcast, false)); + let json = catalog.json.tracks.iter().map(|(name, c)| (name, &c.broadcast, false)); + let binary = catalog + .binary + .tracks + .iter() + .map(|(name, c)| (name, &c.broadcast, false)); + + let mut listed = HashSet::new(); + let mut changes = Vec::new(); + for (name, broadcast, pacing) in video.chain(audio).chain(text).chain(json).chain(binary) { + if let Some(broadcast) = broadcast { + anyhow::bail!( + "track `{name}` is served from broadcast `{broadcast}`; an archive records one broadcast" + ); + } + if !listed.insert(name.clone()) || self.listed.contains(name) { + continue; + } + anyhow::ensure!( + self.enrolled.insert(name.clone()), + "track `{name}` returned to the catalog after it was dropped; the recording cannot resume it" + ); + changes.push(match pacing { + true => Change::Pacing(name.clone()), + false => Change::Track(name.clone()), + }); + } + + let mut dropped: Vec<_> = self.listed.difference(&listed).cloned().collect(); + dropped.sort(); + changes.extend(dropped.into_iter().map(Change::Remove)); + self.listed = listed; + Ok(changes) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use hang::catalog::{AudioCodec, AudioConfig, JsonConfig, Mode}; + use moq_net::{Timescale, Timestamp, group, track}; + + fn audio() -> AudioConfig { + AudioConfig::new(AudioCodec::Opus, 48_000, 2) + } + + /// Renditions pace, data tracks don't, and a later snapshot only adds what is new. + #[test] + fn renditions_pace_and_the_rest_follow() { + let mut tracks = Tracks::default(); + let mut catalog = hang::Catalog::default(); + catalog.audio.renditions.insert("audio".into(), audio()); + catalog + .json + .tracks + .insert("chat".into(), JsonConfig::new(Mode::Snapshot)); + + assert_eq!( + tracks.update(&catalog).unwrap(), + [Change::Pacing("audio".into()), Change::Track("chat".into())] + ); + + catalog.audio.renditions.insert("audio2".into(), audio()); + assert_eq!(tracks.update(&catalog).unwrap(), [Change::Pacing("audio2".into())]); + assert_eq!(tracks.update(&catalog).unwrap(), []); + } + + /// A dropped rendition stops pacing, and coming back is refused rather than silently lost. + #[test] + fn a_dropped_track_is_removed_for_good() { + let mut tracks = Tracks::default(); + let mut catalog = hang::Catalog::default(); + catalog.audio.renditions.insert("audio".into(), audio()); + tracks.update(&catalog).unwrap(); + + let removed = catalog.audio.renditions.remove("audio").unwrap(); + assert_eq!(tracks.update(&catalog).unwrap(), [Change::Remove("audio".into())]); + + catalog.audio.renditions.insert("audio".into(), removed); + let err = tracks.update(&catalog).unwrap_err().to_string(); + assert!(err.contains("cannot resume"), "{err}"); + } + + /// A rendition living in a sibling broadcast would be missing from the recording. + #[test] + fn another_broadcast_is_refused() { + let mut rendition = audio(); + rendition.broadcast = Some(moq_net::path::RelativeOwned::new("./source")); + let mut catalog = hang::Catalog::default(); + catalog.audio.renditions.insert("audio".into(), rendition); + + let err = Tracks::default().update(&catalog).unwrap_err().to_string(); + assert!(err.contains("broadcast `source`"), "{err}"); + } + + /// What `export archive` records, `import archive` serves back group for group, and the + /// export ends cleanly with its broadcast. + #[tokio::test] + async fn a_recording_replays_its_groups() { + // `open` reads the credential variables. + let _env = crate::test_env::EnvGuard::clear(&[]); + let dir = tempfile::tempdir().unwrap(); + let url = Url::from_directory_path(dir.path()).unwrap(); + + let origin = moq_tokio::origin::spawn(); + let mut broadcast = origin.create_broadcast("live.hang").unwrap(); + let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let info = track::Info::default() + .with_timescale(Timescale::MILLI) + .with_max_age(Duration::from_secs(3600)); + let track = broadcast.create_track("audio", info).unwrap(); + catalog + .mutate(|catalog| { + catalog.audio.renditions.insert("audio".into(), audio()); + }) + .unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let args = ExportArgs { + store: url.clone(), + retention: None, + retention_grace: None, + }; + let recording = tokio::spawn(export(origin.consume(), "live.hang".into(), CatalogFormat::Hang, args)); + + for sequence in 0..3 { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + for offset in [0, 500] { + let timestamp = Timestamp::from_millis(sequence * 1000 + offset).unwrap(); + group.write_frame(timestamp, format!("{sequence}+{offset}")).unwrap(); + } + group.finish().unwrap(); + } + + // Finish only once the rendition is enrolled, which writes its `.info`. + tokio::time::timeout(Duration::from_secs(10), async { + while !dir.path().join("audio/.info").exists() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the rendition is enrolled"); + track.finish().unwrap(); + catalog.finish().unwrap(); + broadcast.finish(); + + tokio::time::timeout(Duration::from_secs(10), recording) + .await + .expect("the export ends with its broadcast") + .unwrap() + .expect("the recording succeeds"); + + let replay = moq_tokio::origin::spawn(); + let args = ImportArgs { + store: url, + follow: None, + }; + let serving = tokio::spawn(import(replay.clone(), "replay.hang".into(), args)); + + let consumer = replay.consume().routed_broadcast("replay.hang").await.unwrap(); + let audio = consumer.track("audio").unwrap(); + for sequence in 0..3 { + let mut group = audio.fetch_group(sequence, None).await.unwrap(); + for offset in [0, 500] { + let frame = group.read_frame().await.unwrap().expect("a frame"); + assert_eq!(frame.timestamp.as_millis(), u128::from(sequence * 1000 + offset)); + assert_eq!(frame.payload, format!("{sequence}+{offset}")); + } + assert!(group.read_frame().await.unwrap().is_none()); + } + + serving.abort(); + } +} diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 2e5cff514d..27d87b0f6d 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -18,7 +18,7 @@ //! conditional on the subcommand. //! - The endpoint is one subcommand: a container format (`ts`, `fmp4`, ... read //! from stdin on import, written to stdout on export) or a gateway (`hls`, -//! `rtmp`, `srt`, `rtc`). Exactly one per stage, so "which endpoint" is +//! `rtmp`, `srt`, `rtc`, `archive`). Exactly one per stage, so "which endpoint" is //! unambiguous and there's no silently-ignored flag. //! - `--` starts another stage on the same Origin and the same MoQ attachment, so //! one process can bridge several broadcasts (or both directions at once). Usage @@ -761,6 +761,8 @@ pub enum ImportSource { Srt(crate::srt::Args), /// WebRTC: WHEP client pulling a remote (`--connect`) or WHIP server accepting publishes (`--listen`). Rtc(crate::rtc::Args), + /// Replay a recording from an object store, serving its groups on demand. + Archive(crate::archive::ImportArgs), /// Capture a local source (camera, display, window, app, microphone) and /// encode natively. Run `moq devices` to list them. #[cfg(feature = "capture")] @@ -854,6 +856,8 @@ pub enum ExportSink { Srt(crate::srt::Args), /// WebRTC: WHIP client pushing to a remote (`--connect`) or WHEP server serving plays (`--listen`). Rtc(crate::rtc::Args), + /// Record the broadcast into an object store until it ends. + Archive(crate::archive::ExportArgs), } impl ExportSink { diff --git a/rs/moq-cli/src/main.rs b/rs/moq-cli/src/main.rs index 27ee260451..018c7bbb3a 100644 --- a/rs/moq-cli/src/main.rs +++ b/rs/moq-cli/src/main.rs @@ -4,6 +4,7 @@ //! grammar; this module orchestrates the shared Origin and spawns the MoQ side //! plus every stage's endpoint. +mod archive; mod args; mod auth; mod complete; @@ -638,6 +639,11 @@ fn spawn_import( tasks.spawn(rtc::connect_import(target(name), url)); } } + ImportSource::Archive(args) => { + // A replay serves the retention the recording was made with. + anyhow::ensure!(max_age.is_none(), "`--max-age` does not apply to `import archive`"); + tasks.spawn(archive::import(origin.clone(), name, args)); + } #[cfg(feature = "capture")] ImportSource::Capture(capture) => { warn_if_missing_format(&name); @@ -717,6 +723,18 @@ fn spawn_export( tasks.spawn(rtc::connect_export(origin.consume(), url, name)); } } + ExportSink::Archive(args) => { + anyhow::ensure!( + export.select.is_empty(), + "`export archive` records every rendition; drop the selection flags" + ); + let format = export + .catalog_format + .map(Into::into) + .or_else(|| moq_mux::catalog::CatalogFormat::detect(&name)) + .unwrap_or_default(); + tasks.spawn(archive::export(origin.consume(), name, format, args)); + } _ => unreachable!("container formats are handled by stdout_format above"), } } diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 0680d83f8b..86cb0ae3a4 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -105,6 +105,14 @@ pub struct SelectArgs { } impl SelectArgs { + /// Whether no selection flag was given. + pub(crate) fn is_empty(&self) -> bool { + self.video_name.is_none() + && self.video_codec.is_none() + && self.audio_name.is_none() + && self.audio_codec.is_none() + } + /// Build the rendition selection shared by stdout exports and native playback. /// /// `force` takes the place of `--video-codec`, for a sink whose format implies From 966e7fa7971e012debaec8ac6b587d99ed1e1af3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 06:53:21 -0700 Subject: [PATCH 3/3] fix(cli): hold archive segment commits across a catalog snapshot `export archive` enrolled each rendition while the writer was already committing. A buffered opening rendition could close a segment before the rest of that snapshot was registered, so those groups never entered the record. `Control::reserve` withholds commits until the snapshot's tracks are enrolled. Also sort the workspace `moq-archive` dependency, which failed `cargo sort`. Co-Authored-By: Grok 4.7 --- Cargo.lock | 1 + Cargo.toml | 2 +- rs/moq-archive/src/writer.rs | 15 +++++- rs/moq-cli/Cargo.toml | 2 + rs/moq-cli/src/archive.rs | 91 ++++++++++++++++++++++++++++++++++++ 5 files changed, 109 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0c1fc204ec..cf56ca047c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4322,6 +4322,7 @@ dependencies = [ "moq-audio", "moq-auth", "moq-hls", + "moq-json", "moq-msf", "moq-mux", "moq-relay", diff --git a/Cargo.toml b/Cargo.toml index f6b12a60e4..0a72c53233 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -139,8 +139,8 @@ loom = { version = "0.7.2", features = ["futures"] } # DNS-SD advertisement and browsing for LAN peer discovery (moq-tokio's `mdns` feature). # `async` awaits the event channel instead of blocking a thread on it. mdns-sd = { version = "0.21", features = ["async"] } -moq-audio = { version = "0.1.2", path = "rs/moq-audio", default-features = false } moq-archive = { version = "0.0.3", path = "rs/moq-archive" } +moq-audio = { version = "0.1.2", path = "rs/moq-audio", default-features = false } moq-auth = { version = "0.1.1", path = "rs/moq-auth" } moq-binary = { version = "0.1.2", path = "rs/moq-binary" } moq-flate = { version = "0.2.0", path = "rs/moq-flate" } diff --git a/rs/moq-archive/src/writer.rs b/rs/moq-archive/src/writer.rs index 13ac32294a..3c545f8a67 100644 --- a/rs/moq-archive/src/writer.rs +++ b/rs/moq-archive/src/writer.rs @@ -31,7 +31,7 @@ use futures::future::BoxFuture; use futures::stream::FuturesUnordered; use futures::{FutureExt, StreamExt}; use hang::timeline::Record; -use moq_mux::timeline::{self, Deferred, DeferredDrain, Pending, Recorder}; +use moq_mux::timeline::{self, Deferred, DeferredDrain, Pending, Recorder, Reserved}; use moq_net::{Timescale, Timestamp, broadcast, group, track}; use object_store::ObjectStore; use tokio::sync::{mpsc, watch}; @@ -116,6 +116,11 @@ impl Clone for Control { } } +/// Withholds segment commits until dropped. +pub struct Reservation { + _inner: Reserved, +} + struct Shared { store: Store, source: broadcast::Consumer, @@ -340,6 +345,14 @@ impl Control { self.send(Command::Poke) } + /// Hold segment commits back until this guard drops, so a batch of tracks can enroll first. + #[must_use = "dropping the reservation releases segment commits"] + pub fn reserve(&self) -> Reservation { + Reservation { + _inner: self.deferred.reserve(), + } + } + /// Stop recording `name`, dropping its incomplete groups. The name cannot be enrolled again. pub fn remove(&self, name: &str) -> Result<()> { self.send(Command::Remove(name.to_string())) diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index f5830ff03e..7ee2a123d2 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -133,6 +133,8 @@ winit = { version = "0.30.13", optional = true } sd-notify = { workspace = true } [dev-dependencies] +# Decodes the timeline the archive round-trip test checks. +moq-json = { workspace = true } # `test_relay` stands up a real relay for the `fetch` tests to read through. moq-relay = { path = "../moq-relay", default-features = false, features = ["test-support"] } tempfile = { workspace = true } diff --git a/rs/moq-cli/src/archive.rs b/rs/moq-cli/src/archive.rs index f8eec6949c..2689096655 100644 --- a/rs/moq-cli/src/archive.rs +++ b/rs/moq-cli/src/archive.rs @@ -154,6 +154,10 @@ async fn enroll( ) -> anyhow::Result<()> { let mut tracks = Tracks::default(); while let Some(snapshot) = catalog.next().await? { + // One rendition's subscription can outrun the rest of this snapshot. Without the + // hold, the writer closes a segment from that rendition alone and the others never + // enter the record. + let _hold = control.reserve(); for change in tracks.update(&snapshot)? { match change { Change::Pacing(name) => control.pacing_track(&name).await?, @@ -380,4 +384,91 @@ mod tests { serving.abort(); } + + /// Renditions already live in the opening catalog all land in its first segment. + /// + /// Groups are published before export starts, so the writer can drain the first rendition + /// while the next subscription is still in flight. + #[tokio::test] + async fn an_opening_snapshot_records_every_rendition() { + let _env = crate::test_env::EnvGuard::clear(&[]); + let dir = tempfile::tempdir().unwrap(); + let url = Url::from_directory_path(dir.path()).unwrap(); + + let origin = moq_tokio::origin::spawn(); + let mut broadcast = origin.create_broadcast("live.hang").unwrap(); + let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let info = track::Info::default() + .with_timescale(Timescale::MILLI) + .with_max_age(Duration::from_secs(3600)); + let first = broadcast.create_track("audio", info.clone()).unwrap(); + let second = broadcast.create_track("audio2", info).unwrap(); + for track in [&first, &second] { + for sequence in 0..3 { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + for offset in [0, 500] { + let timestamp = Timestamp::from_millis(sequence * 1000 + offset).unwrap(); + group.write_frame(timestamp, format!("{sequence}+{offset}")).unwrap(); + } + group.finish().unwrap(); + } + } + catalog + .mutate(|catalog| { + catalog.audio.renditions.insert("audio".into(), audio()); + catalog.audio.renditions.insert("audio2".into(), audio()); + }) + .unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let args = ExportArgs { + store: url.clone(), + retention: None, + retention_grace: None, + }; + let recording = tokio::spawn(export(origin.consume(), "live.hang".into(), CatalogFormat::Hang, args)); + + tokio::time::timeout(Duration::from_secs(10), async { + while !dir.path().join("audio/.info").exists() || !dir.path().join("audio2/.info").exists() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("both renditions are enrolled"); + first.finish().unwrap(); + second.finish().unwrap(); + catalog.finish().unwrap(); + broadcast.finish(); + + tokio::time::timeout(Duration::from_secs(10), recording) + .await + .expect("the export ends with its broadcast") + .unwrap() + .expect("the recording succeeds"); + + let store = super::open(&url).unwrap(); + let object = store + .get_segments(hang::timeline::DEFAULT_NAME, 0) + .await + .expect("segment 0"); + let config = moq_json::window::ConsumerConfig::default().with_compression(true); + let mut decoder = moq_json::window::Decoder::::new(config); + for stored in object.groups { + let mut group = decoder.group(); + for frame in stored.frames { + group.decode(&frame.payload).unwrap(); + } + } + let mut opening = None; + while let Some(event) = decoder.next_event() { + if let moq_json::window::Event::Push { value, .. } = event + && value.segment == 0 + { + opening = Some(value); + } + } + let opening = opening.expect("segment 0 has a record"); + assert!(opening.tracks.contains_key("audio"), "{opening:?}"); + assert!(opening.tracks.contains_key("audio2"), "{opening:?}"); + } }