feat(cli): record and replay archives with export archive / import archive

Cargo.toml

@@ -139,6 +139,7 @@ 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-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" }

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 <url>` | Pull a remote HLS playlist. |
 | `import` | `rtmp`, `srt`, `rtc` | Accept pushes (`--listen`) or pull from a remote (`--connect`). |
+| `import` | `archive <url>` | 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 <url>` | 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` | `<track>` | 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

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

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<S> Clone for Control<S> {
 	}
 }
 
+/// Withholds segment commits until dropped.
+pub struct Reservation {
+	_inner: Reserved,
+}
+
 struct Shared<S> {
 	store: Store<S>,
 	source: broadcast::Consumer,
@@ -340,6 +345,14 @@ impl<S: ObjectStore> Control<S> {
 		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()))

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,8 @@ 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" }
+# 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 }

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 <MoQ side> <import|export> <endpoint>`. The MoQ side (before the verb) dials with `--connect <url>`, self-hosts QUIC/WebTransport with `--listen <addr>`, or self-hosts raw qmux with `--listen-tcp-bind <addr>` / `--listen-unix-bind <path>` (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 <MoQ side> play`.
+`moq-cli` routes endpoints onto a shared MoQ Origin: `moq <MoQ side> <import|export> <endpoint>`. The MoQ side (before the verb) dials with `--connect <url>`, self-hosts QUIC/WebTransport with `--listen <addr>`, or self-hosts raw qmux with `--listen-tcp-bind <addr>` / `--listen-unix-bind <path>` (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 <MoQ side> play`.
 
 Separate additional stages with `--` to bridge several broadcasts (or both directions) over one connection, each naming its own `--broadcast`: