From 20d808162dca3e8cbd434bc107d48d7c35af9d03 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 19:58:34 -0800 Subject: [PATCH 01/17] chore(quest): claim quest/m0/recording.md Quest-Claim: 6fcc1200df9060d7 From c4f2cfedd9c39e201b1c446dc9317bce936477ae Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 22:06:15 -0800 Subject: [PATCH 02/17] feat(studio): record the program and each source to MP4 Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + Cargo.toml | 1 + README.md | 28 +- justfile | 2 +- rs/moq-scene/src/publish.rs | 76 +++- rs/moq-studio/Cargo.toml | 3 + rs/moq-studio/src/engine.rs | 309 ++++++++++----- rs/moq-studio/src/main.rs | 28 +- rs/moq-studio/src/record.rs | 756 ++++++++++++++++++++++++++++++++++++ rs/moq-studio/src/ui.rs | 73 +++- 10 files changed, 1170 insertions(+), 107 deletions(-) create mode 100644 rs/moq-studio/src/record.rs diff --git a/Cargo.lock b/Cargo.lock index 9395c63f..5758699c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3460,6 +3460,7 @@ dependencies = [ "egui", "egui_dock", "hang", + "humantime", "kio", "moq-audio", "moq-auth", diff --git a/Cargo.toml b/Cargo.toml index 420e5bfa..e3f3bf5d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,7 @@ moq-video = { path = "moq/rs/moq-video", default-features = false } anyhow = "1" bytes = "1" +humantime = "2.4" humantime-serde = "1.1.1" pollster = "1.0" serde = { version = "1", features = ["derive"] } diff --git a/README.md b/README.md index 36d3391b..202b2662 100644 --- a/README.md +++ b/README.md @@ -15,8 +15,10 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ - `show.toml` is the scene collection. It is created if it does not exist, and changes are saved to it a second after you stop editing, with no save step. -- Without `--connect`, the studio composes and previews but publishes nothing, - and broadcast sources stay offline. +- `--recordings ` is where recordings are saved, by default beside + `show.toml`. +- Without `--connect`, the studio composes, previews, and records but + publishes nothing, and broadcast sources stay offline. - `capture` adds camera, display, window, microphone, and system-audio sources. On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` is the software H.264 fallback for a machine without a hardware encoder. @@ -39,7 +41,7 @@ Every panel is a tab you can drag, split, or stack. | Devices | List this machine's cameras, displays, windows, and audio inputs, and add one as a source. | | Filters | Add, order, and set the filters of the selected source or item, and copy or paste them. | | Audio | Set each audio source's gain and mute, and watch its meter and the program's. | -| Output | See the broadcast path, whether the program is encoding, and the encoder's work. | +| Output | Start and stop recording, and see the broadcast path, whether the program is encoding, and the encoder's work. | | Health | See how the stream reaches viewers, track by track, and the cause of each problem. | A new source, from the Sources or Devices panel, is placed over whatever the @@ -198,7 +200,8 @@ the hang draft specifies source listings: A captured source encodes as H.264 for as long as it is in the collection, whether or not anyone watches, so a late viewer can fetch its recent past; the -relay uploads only what someone subscribes to. A capture that changes size or +relay uploads only what someone subscribes to. Without `--connect`, it encodes +only while a recording reads it. A capture that changes size or color space moves to a new track, which its catalog entry follows. Images and colors publish once, as the image file and as a one-pixel PNG. Removing or editing a source ends its track. A track carries the plain capture, before any @@ -216,6 +219,23 @@ past the cap fail rather than the program. When the program encodes in software, a source takes hardware where it can and software otherwise. +### Recording + +**Start recording** in the Output panel saves the program to +` .mp4` in the `--recordings` directory, live +or not. With **each source too**, every camera, display, and window that +publishes a track of its own records into a file of its own beside it, +` .mp4`, holding the capture before +any filter. A source added while recording starts its file then, and one +removed finishes its file then. A recording reads the broadcast the way a +viewer does, so it shares the live encoders and starts and stops without +touching what the relay serves. Without `--connect`, or with `--on-demand`, +the program encodes only while something records or watches it. + +Each file plays while it grows, so a crash leaves a playable fragmented MP4, +and becomes a regular MP4 when the recording stops or the studio quits. A +file nothing reached before the recording stopped is removed. + ### Source frames Sources make the opposite trip. A frame already on the GPU is drawn where it diff --git a/justfile b/justfile index 2bd9ae7b..e6b12976 100644 --- a/justfile +++ b/justfile @@ -48,7 +48,7 @@ render-sw: icd=("$mesa"/share/vulkan/icd.d/lvp_icd.*.json) export LD_LIBRARY_PATH="$loader/lib" VK_DRIVER_FILES="${icd[0]}" cargo nextest run --locked -p moq-scene --lib --run-ignored only -E 'test(/^compositor::tests::/)' - cargo nextest run --locked -p moq-studio --features openh264 --run-ignored only -E 'test(=engine::tests::the_program_reaches_a_relay)' + cargo nextest run --locked -p moq-studio --features openh264 --run-ignored only -E 'test(/^engine::tests::the_program_/)' # Move the moq submodule to moq's latest main (or a given commit) and refresh Cargo.lock. upgrade-moq rev="origin/main": diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 3fbb6b82..a2f5f175 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -8,7 +8,8 @@ //! be played, or packaged for HLS, as a variant of the program. //! //! A source encodes while it runs, whatever the demand, so a late viewer finds -//! its recent past; MoQ uploads only what a subscriber pulls. +//! its recent past; MoQ uploads only what a subscriber pulls. On +//! [demand](Publish::on_demand), it encodes only while subscribed. use std::collections::BTreeMap; use std::sync::{Arc, Mutex}; @@ -57,6 +58,7 @@ pub struct Publish { catalog: Catalog, bandwidth: moq_net::bandwidth::Allocator, kind: moq_video::encode::Kind, + on_demand: bool, owners: Arc>, } @@ -86,10 +88,19 @@ impl Publish { catalog, bandwidth, kind, + on_demand: false, owners: Default::default(), } } + /// Encode each source's video only while someone subscribes to it, for a + /// studio whose only readers are its own recordings, which never fetch a + /// past they were not reading. + pub fn on_demand(mut self) -> Self { + self.on_demand = true; + self + } + /// Claim `name`'s entry in the studio catalog for a source that is opening, /// taking it from any older source of that name. pub(crate) fn claim(&self, name: &str) -> Claim { @@ -138,6 +149,7 @@ impl Drop for Claim { pub(crate) struct Listing<'a> { claim: &'a Claim, kind: moq_video::encode::Kind, + on_demand: bool, broadcast: moq_net::broadcast::Producer, studio: Catalog, catalog: moq_mux::catalog::Producer, @@ -158,6 +170,7 @@ impl<'a> Listing<'a> { Ok(Self { claim, kind: publish.kind.clone(), + on_demand: publish.on_demand, broadcast: publish.broadcast.clone(), studio: publish.catalog.clone(), catalog, @@ -211,6 +224,8 @@ pub(crate) struct Video { /// Opened on the first frame, and reopened when the picture's size or color /// space changes, which moves the rendition to a new track. encoder: Option, + /// Encodes only while the track is subscribed. + on_demand: bool, clock: moq_mux::Clock, /// Where the capture's private timeline starts on the studio clock, in /// microseconds, taken at the first frame. @@ -246,12 +261,14 @@ impl Video { producer: Some(producer), config, encoder: None, + on_demand: listing.on_demand, clock: listing.studio.clock(), epoch: None, }) } - /// Encode and publish `frame`, restamping it onto the studio clock. + /// Encode and publish `frame`, restamping it onto the studio clock, unless + /// it is on demand and nobody subscribes. #[cfg_attr( not(feature = "capture"), allow(dead_code, reason = "only capture sources encode video") @@ -263,6 +280,17 @@ impl Video { .get_or_insert_with(|| self.clock.now().as_micros().saturating_sub(captured)); frame.timestamp = moq_net::Timestamp::from_micros((epoch + captured) as u64).map_err(moq_video::Error::from)?; + let producer = self.producer.as_mut().expect("open until it ends"); + if self.on_demand && producer.demand().wanted().is_none() { + // The next subscriber starts on a fresh encoder's keyframe, and the + // session is free meanwhile; the gap is marked so nobody reads across it. + if self.encoder.is_some() { + self.drain()?; + self.producer.as_mut().expect("open until it ends").discontinuity()?; + } + return Ok(()); + } + let size = encodable(frame.size()); let color = frame.surface.color(); if size != self.config.size() || color.is_some_and(|color| Some(color) != self.config.color) { @@ -543,6 +571,50 @@ mod tests { .await; } + /// An on-demand source encodes only while its track is subscribed, holding + /// no encoder between subscribers. + #[tokio::test] + async fn an_on_demand_video_encodes_while_subscribed() { + local(async { + let (publish, catalog, consumer) = studio(); + let publish = publish.on_demand(); + let claim = publish.claim("camera"); + let listing = Listing::new(&claim, &publish).unwrap(); + let framerate = moq_video::FrameRate::new(30, 1).unwrap(); + let size = moq_video::Size::new(64, 48); + + tokio::select! { + result = listing.mirror() => panic!("the mirror ended: {result:?}"), + () = async { + let mut video = Video::new(&listing, size, framerate, None).await.unwrap(); + video.encode(&mut gray(size, 0)).unwrap(); + assert!(video.encoder.is_none(), "nothing encodes unsubscribed"); + + // Advertised before anything encodes, so a recording can find it. + let rendition = |sources: &BTreeMap| { + sources.get("camera").is_some_and(|camera| !camera.video.renditions.is_empty()) + }; + let snapshot = listed(&catalog, rendition).await; + let name = snapshot.ext.studio.sources["camera"].video.renditions.keys().next().unwrap().clone(); + let mut track = consumer.track(&name).unwrap().subscribe(None).await.unwrap(); + video.encode(&mut gray(size, 33)).unwrap(); + assert!(video.encoder.is_some(), "a subscriber gets pictures"); + let mut group = track.recv_group().await.unwrap().expect("a group"); + assert!(group.read_frame().await.unwrap().is_some()); + + drop(group); + drop(track); + for _ in 0..100 { + tokio::task::yield_now().await; + } + video.encode(&mut gray(size, 66)).unwrap(); + assert!(video.encoder.is_none(), "the encoder closes once nobody subscribes"); + } => {} + } + }) + .await; + } + /// An edited source's replacement owns the listing from the moment it /// opens, even when the old source's driver only gets to list itself later, /// and the old one's end leaves the replacement listed. diff --git a/rs/moq-studio/Cargo.toml b/rs/moq-studio/Cargo.toml index 52bd1efc..9b9a994b 100644 --- a/rs/moq-studio/Cargo.toml +++ b/rs/moq-studio/Cargo.toml @@ -43,6 +43,9 @@ egui = "0.36" # Docked, rearrangeable panels. 0.21 is the newest release and pins egui 0.36. egui_dock = "0.21" hang = { workspace = true } +# Stamps a recording's file name with when it started. Already in the graph +# through moq-scene's humantime-serde. +humantime = { workspace = true } kio = { workspace = true, features = ["async", "runtime"] } moq-audio = { workspace = true, features = ["bus"] } moq-json = { workspace = true } diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index d4d466b4..0f917e5f 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -1,12 +1,14 @@ //! Everything behind the window: the GPU, the sources, the program clock that -//! draws the canvas and mixes the audio, and the broadcast it publishes: the -//! program, and each local source's own track beside it. +//! draws the canvas and mixes the audio, and the broadcast it encodes into: +//! the program, and each local source's own track beside it. The relay serves +//! that broadcast, and recordings read it. //! //! Threads: //! - **program** draws preview and program once per canvas frame and hands the //! program to the encoder, stamped with the program clock. //! - **audio** reads the program bus on the same clock and publishes it. //! - **encode** waits for each program copy and encodes it. +//! - Recordings write their files on the tokio runtime. //! - **source-***: one per running source, since a capture stream may not //! leave the thread that opened it. //! - **autosave** writes the collection and its history to their files once @@ -26,7 +28,7 @@ use moq_scene::{Collection, Compositor, Draft, History, wgpu}; use moq_video::render::export::{Exporter, Pending, Route}; use crate::autosave::{self, Autosave}; -use crate::health; +use crate::{health, record}; /// The GPU everything shares: sources upload to it, the compositor draws on /// it, the window presents from it, and the encoder reads the program from it. @@ -180,7 +182,8 @@ pub struct Stats { pub encoded: AtomicU64, /// Microseconds the last encode took. pub encode_us: AtomicU64, - /// Whether any viewer is subscribed to the video. + /// Whether anyone subscribes to the program video: a viewer through the + /// relay, or a recording. pub watched: AtomicBool, /// The encoder backend, once one opened. pub encoder: Mutex>, @@ -195,20 +198,27 @@ impl Stats { } } -/// Where the program goes. +/// The relay the program goes live on. pub struct Output { /// The broadcast path on the relay. pub path: String, /// Whether the program video is copied and encoded only while someone - /// watches, rather than all the time. + /// watches or records it, rather than all the time. pub on_demand: bool, /// The relay connection, which reconnects on its own and dials only while /// held. The Health panel reads its state. _connection: moq_tokio::Connection, +} + +/// The program's broadcast, open for as long as the engine runs. +struct Program { + /// Where the broadcast is announced: the relay connection publishes it, + /// and recordings read it. + origin: moq_net::origin::Producer, publication: moq_net::origin::Publication, } -impl Drop for Output { +impl Drop for Program { fn drop(&mut self) { self.publication.close(); } @@ -224,6 +234,10 @@ pub struct Engine { pub bus: moq_audio::bus::Bus, /// `None` when running without a relay. pub output: Option, + /// Records the program, or says why it cannot: without a relay, a machine + /// that cannot encode the canvas still composes and previews. + pub recorder: Result, + _program: Option, /// Keeps broadcast sources' origin open when no connection holds it. _origin: Option, } @@ -238,8 +252,9 @@ const AUDIO_PERIOD: Duration = Duration::from_millis(10); impl Engine { /// Open the `loaded` collection and history on `gpu`, start the program /// clock, and save both to `autosave`'s files as they change. - /// With `connect`, publish the program at `broadcast` on that relay, - /// encoding its video only while someone watches when `on_demand`. + /// The program is encoded into a broadcast at `broadcast` while anyone + /// wants it: a recording into `recordings`, or with `connect`, that relay, + /// which wants it all the time unless `on_demand`. /// /// Call inside the tokio runtime that serves the relay connection. pub async fn start( @@ -249,6 +264,7 @@ impl Engine { connect: Option<(moq_tokio::Client, url::Url)>, broadcast: &str, on_demand: bool, + recordings: &Path, ) -> anyhow::Result> { let bus = moq_audio::bus::Bus::new(Default::default())?; let subscribe = moq_tokio::origin::spawn(); @@ -260,13 +276,34 @@ impl Engine { } = loaded; let canvas = collection.canvas; - let (output, origin, publish) = match connect { + let path = moq_net::Path::new(broadcast).with_context(|| format!("cannot publish at `{broadcast}`"))?; + let origin = moq_tokio::origin::spawn(); + let (relay, subscribe) = match connect { Some((client, url)) => { - let (output, publish) = publish(&gpu, client, url, subscribe, broadcast, canvas, on_demand).await?; - (Some(output), None, Some(publish)) + tracing::info!(url = %moq_tokio::RedactedUrl::new(&url), broadcast = %path, "connecting"); + let reports = subscribe.consume(); + let connection = client.with_publisher(&origin).with_subscriber(subscribe).connect(url); + (Some(Relay { connection, reports }), None) } - None => (None, Some(subscribe), None), + None => (None, Some(subscribe)), }; + let (program, publish, recorder) = match publish(&gpu, origin, &path, canvas, relay.as_ref()).await { + Ok((program, publish)) => { + let broadcast = program.publication.broadcast().consume(); + let source = moq_mux::Source::new(program.origin.consume(), broadcast); + let recorder = record::Recorder::new(source, recordings, autosave.path())?; + (Some(program), Some(publish), Ok(recorder)) + } + // Without a relay the studio still composes and previews; only a + // recording needs the encoder. + Err(err) if relay.is_none() => (None, None, Err(format!("{err:#}"))), + Err(err) => return Err(err), + }; + let output = relay.map(|relay| Output { + path: path.to_string(), + on_demand, + _connection: relay.connection, + }); // Absolute, so a copied image names the same file in another collection. let dir = autosave.path().parent().filter(|dir| !dir.as_os_str().is_empty()); @@ -297,7 +334,9 @@ impl Engine { health: Mutex::default(), bus, output, - _origin: origin, + recorder, + _program: program, + _origin: subscribe, }); let (video, audio, clock, publisher) = match publish { @@ -324,7 +363,7 @@ impl Engine { // its encode session even while unwatched, and a source past the // GPU's session cap is the one that fails. open.await.context("the encode thread stopped")??; - (Some((frames, exporter)), Some(audio), Some(clock), Some(health)) + (Some((frames, exporter)), Some(audio), Some(clock), health) } None => (None, None, None, None), }; @@ -386,7 +425,6 @@ impl Engine { fn local(&self, now: Instant) -> health::Local { let stats = &self.stats; let watched = stats.watched.load(Ordering::Relaxed); - let on_demand = self.output.as_ref().is_some_and(|output| output.on_demand); let stalled = { let state = self.state.lock().unwrap(); let program = state @@ -413,7 +451,7 @@ impl Engine { rendered: stats.rendered.load(Ordering::Relaxed), dropped: stats.dropped.load(Ordering::Relaxed), encoded: stats.encoded.load(Ordering::Relaxed), - encoding: self.output.is_some() && (!on_demand || watched), + encoding: self.recorder.is_ok() && (self.always() || watched), watched, encoder: stats.encoder.lock().unwrap().clone(), failed: stats.error.lock().unwrap().clone(), @@ -421,9 +459,15 @@ impl Engine { } } + /// Whether the program video encodes whether or not anyone wants it: live on + /// a relay, and not on demand. + fn always(&self) -> bool { + self.output.as_ref().is_some_and(|output| !output.on_demand) + } + /// Draw a canvas frame per tick of the program clock, and hand the program - /// to the encoder, stamped `epoch` past `start`: every frame, or only while - /// anyone watches when the output is on demand. + /// to the encoder, stamped `epoch` past `start`: every frame while live and + /// not on demand, otherwise only while anyone watches or records it. /// /// Drawing outlives the encoder: a failed output stops copying the program, /// and the preview and program keep moving. @@ -437,7 +481,7 @@ impl Engine { let period = Duration::from_secs(1) / canvas.framerate; let mut clock = kio::clock::interval_at(start, period); clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); - let on_demand = self.output.as_ref().is_some_and(|output| output.on_demand); + let always = self.always(); let mut encoding = false; loop { @@ -459,7 +503,7 @@ impl Engine { self.stats.rendered.fetch_add(1, Ordering::Relaxed); let Some((sender, exporter)) = &mut video else { continue }; - let wanted = !on_demand || self.stats.watched.load(Ordering::Relaxed); + let wanted = always || self.stats.watched.load(Ordering::Relaxed); let was = std::mem::replace(&mut encoding, wanted); if !encoding { // Behind every frame already sent, so the break lands after them. @@ -534,7 +578,14 @@ impl Engine { } } -/// The encoders behind an [`Output`], handed to their threads. +/// The relay connection the program goes live on, and the origin its viewers' +/// reports and the relay's stats arrive on. +struct Relay { + connection: moq_tokio::Connection, + reports: moq_net::origin::Consumer, +} + +/// The encoders behind the program's broadcast, handed to their threads. struct Publish { video: moq_video::encode::Producer, config: moq_video::encode::Config, @@ -543,41 +594,35 @@ struct Publish { /// The catalog's clock, which the codec producers write timestamps /// against verbatim. clock: moq_mux::Clock, - /// What the health sampler grades. - health: health::Publisher, + /// What the health sampler grades, while live on a relay. + health: Option, /// Where each local source publishes its own track. sources: moq_scene::publish::Publish, } -/// Announce the program broadcast, which also carries each local source's own -/// track, and connect to the relay, which also carries `subscribe`'s broadcast -/// sources. +/// Announce the program broadcast at `path` on `origin`, which `relay` +/// publishes when there is one. The broadcast also carries each local +/// source's own track. async fn publish( gpu: &Gpu, - client: moq_tokio::Client, - url: url::Url, - subscribe: moq_net::origin::Producer, - path: &str, + origin: moq_net::origin::Producer, + path: &moq_net::Path, canvas: moq_scene::Canvas, - on_demand: bool, -) -> anyhow::Result<(Output, Publish)> { - let origin = moq_tokio::origin::spawn(); - let path = moq_net::Path::new(path).with_context(|| format!("cannot publish at `{path}`"))?; - tracing::info!(url = %moq_tokio::RedactedUrl::new(&url), broadcast = %path, "connecting"); - let reports = subscribe.consume(); - let connection = client.with_publisher(&origin).with_subscriber(subscribe).connect(url); - - let publication = origin.create_broadcast(&path).context("cannot create the broadcast")?; + relay: Option<&Relay>, +) -> anyhow::Result<(Program, Publish)> { + let publication = origin.create_broadcast(path).context("cannot create the broadcast")?; let mut broadcast = publication.broadcast().clone(); // Each track claims its bitrate on the connection's estimate, which the // connection then probes to fund: the bandwidth the stream is granted. - let bandwidth = connection.allocator(); + let bandwidth = match relay { + Some(relay) => relay.connection.allocator(), + None => moq_net::bandwidth::Allocator::unlimited(), + }; let catalog = moq_mux::catalog::Config::default() .with_bandwidth(bandwidth.clone()) .with_catalog(Default::default()); let catalog = moq_scene::publish::Catalog::new(&mut broadcast, catalog)?; let clock = catalog.clock(); - let (echo, viewers) = health::echo(&path)?; let mut config = moq_video::encode::Config::new(canvas.width, canvas.height, canvas.framerate()); // Opens a throwaway encoder now, so a machine that cannot encode the @@ -594,13 +639,26 @@ async fn publish( moq_video::encode::Kind::Named(name) if name == "openh264" => moq_video::encode::Kind::Auto, _ => moq_video::encode::Kind::Hardware, }; - let sources = moq_scene::publish::Publish::new(broadcast.clone(), catalog.clone(), bandwidth, kind); + let mut sources = moq_scene::publish::Publish::new(broadcast.clone(), catalog.clone(), bandwidth, kind); + // Without a relay only a recording reads a source's track, and nothing + // fetches a past it was not recording. + if relay.is_none() { + sources = sources.on_demand(); + } let reserved = catalog.reserve().with_hls_hint(config.gop.duration(config.framerate)); - // While the reservation holds, so the solicitation rides the first catalog. - let mut soliciting = catalog.clone(); - let mut section = soliciting.modify()?; - section.echo = Some(hang::catalog::Echo::new(echo)); - section.commit()?; + // Viewers report back beneath the broadcast, which only a relay carries. + let viewers = match relay { + Some(_) => { + let (echo, viewers) = health::echo(path)?; + // While the reservation holds, so the solicitation rides the first catalog. + let mut soliciting = catalog.clone(); + let mut section = soliciting.modify()?; + section.echo = Some(hang::catalog::Echo::new(echo)); + section.commit()?; + Some(viewers) + } + None => None, + }; let video = moq_video::encode::Producer::new(broadcast.clone(), reserved, rendition)?; let bus = moq_audio::bus::Config::default(); @@ -618,30 +676,29 @@ async fn publish( .announce(moq_net::origin::Route::default().with_epoch(epoch.clone())) .context("cannot announce the broadcast")?; - let feedback = health::Feedback::default(); - let watching = health::Watching::default(); - tokio::spawn(log( - "viewer feedback", - feedback.clone().follow(reports.clone(), viewers), - )); - tokio::spawn(log("relay stats", watching.clone().follow(reports, path.clone()))); - let health = health::Publisher { - observer: hang::health::Observer::new(path.clone(), Some(epoch)), - catalog: stats, - clock: clock.clone(), - monitor: connection.monitor(), - feedback, - watching, - }; + let health = relay.zip(viewers).map(|(relay, viewers)| { + let feedback = health::Feedback::default(); + let watching = health::Watching::default(); + tokio::spawn(log( + "viewer feedback", + feedback.clone().follow(relay.reports.clone(), viewers), + )); + tokio::spawn(log( + "relay stats", + watching.clone().follow(relay.reports.clone(), path.clone()), + )); + health::Publisher { + observer: hang::health::Observer::new(path.clone(), Some(epoch)), + catalog: stats, + clock: clock.clone(), + monitor: relay.connection.monitor(), + feedback, + watching, + } + }); - let output = Output { - path: path.to_string(), - on_demand, - _connection: connection, - publication, - }; Ok(( - output, + Program { origin, publication }, Publish { video, config, @@ -797,23 +854,8 @@ mod tests { /// Bounds each step, so a stall fails rather than hangs. const TIMEOUT: Duration = Duration::from_secs(30); - /// The program a studio publishes reaches a relay as the picture it composed, - /// with its audio alongside. - #[tokio::test(flavor = "multi_thread")] - #[ignore = "needs a Vulkan device and an H.264 encoder; run by just rs render-sw"] - async fn the_program_reaches_a_relay() { - let _ = moq_tokio::crypto::install_default(); - let fixture = moq_relay::test_relay().await.expect("test relay"); - let ready = fixture.relay.ready(); - let relay = fixture.relay.cluster().origin.consume(); - tokio::spawn(fixture.relay.run()); - ready.wait().await.expect("relay ready"); - - let mut client = moq_tokio::connect::Config::default(); - client.tls.fingerprint = vec![fixture.fingerprint.clone()]; - let client = client.init(Default::default()).expect("client"); - - let dir = tempfile::tempdir().unwrap(); + /// A small canvas filled red. + fn red() -> autosave::Loaded { let mut collection = Collection::new(); collection.canvas = moq_scene::Canvas { width: 320, @@ -832,17 +874,98 @@ mod tests { visible: true, filters: Vec::new(), }); + autosave::Loaded { + collection, + history: History::new(), + dropped: 0, + } + } + + /// Record until `encoded` program frames were encoded, and return the + /// handler of every track in each file, by file. + async fn record(engine: &Engine, encoded: u64, sources: bool) -> Vec> { + let recorder = engine.recorder.as_ref().expect("the canvas can be encoded"); + recorder.start(sources).expect("recording starts"); + kio::clock::timeout(TIMEOUT, async { + while engine.stats.encoded.load(Ordering::Relaxed) < encoded { + kio::clock::sleep(Duration::from_millis(100)).await; + } + }) + .await + .expect("the recording encodes the program"); + kio::clock::timeout(TIMEOUT, recorder.finish()) + .await + .expect("the recording finishes"); + + use moq_mux::container::fmp4::mp4_atom::{self, DecodeMaybe}; + let status = recorder.status().expect("a recording"); + status + .files + .iter() + .map(|file| { + assert_eq!(file.finished, Some(record::Finished::Saved), "{}", file.path.display()); + let data = std::fs::read(&file.path).unwrap(); + assert_eq!(file.written, data.len() as u64); + let mut cursor = std::io::Cursor::new(data.as_slice()); + assert!( + moq_mux::container::mp4::repair(&mut cursor).unwrap().is_none(), + "a regular MP4" + ); + let mut cursor = std::io::Cursor::new(data.as_slice()); + let mut handlers = Vec::new(); + while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor).unwrap() { + if let mp4_atom::Any::Moov(moov) = atom { + handlers.extend(moov.trak.iter().map(|trak| trak.mdia.hdlr.handler.to_string())); + } + } + handlers.sort(); + handlers + }) + .collect() + } + + /// Without a relay, a recording encodes the program on its own and saves it + /// as a regular MP4 holding its video and audio. The color source publishes + /// a still, which has no file of its own. + #[tokio::test(flavor = "multi_thread")] + #[ignore = "needs a Vulkan device and an H.264 encoder; run by just rs render-sw"] + async fn the_program_records_without_a_relay() { + let dir = tempfile::tempdir().unwrap(); + let gpu = Gpu::new().await.expect("a GPU"); + let (autosave, _) = Autosave::open(dir.path().join("show.toml")).unwrap(); + let engine = Engine::start(gpu, autosave, red(), None, "studio", false, dir.path()) + .await + .expect("the studio starts"); + assert!(engine.output.is_none()); + assert_eq!(engine.stats.encoded.load(Ordering::Relaxed), 0); + + // Long enough for the first group to close, so the video has a fragment. + let files = record(&engine, 90, true).await; + assert_eq!(files, [["soun", "vide"]]); + } + + /// The program a studio publishes reaches a relay as the picture it composed, + /// with its audio alongside. + #[tokio::test(flavor = "multi_thread")] + #[ignore = "needs a Vulkan device and an H.264 encoder; run by just rs render-sw"] + async fn the_program_reaches_a_relay() { + let _ = moq_tokio::crypto::install_default(); + let fixture = moq_relay::test_relay().await.expect("test relay"); + let ready = fixture.relay.ready(); + let relay = fixture.relay.cluster().origin.consume(); + tokio::spawn(fixture.relay.run()); + ready.wait().await.expect("relay ready"); + + let mut client = moq_tokio::connect::Config::default(); + client.tls.fingerprint = vec![fixture.fingerprint.clone()]; + let client = client.init(Default::default()).expect("client"); + let dir = tempfile::tempdir().unwrap(); let gpu = Gpu::new().await.expect("a GPU"); let (autosave, _) = Autosave::open(dir.path().join("show.toml")).unwrap(); let connect = Some((client, fixture.url.clone())); let on_demand = false; - let loaded = autosave::Loaded { - collection, - history: History::new(), - dropped: 0, - }; - let engine = Engine::start(gpu, autosave, loaded, connect, "studio", on_demand) + let engine = Engine::start(gpu, autosave, red(), connect, "studio", on_demand, dir.path()) .await .expect("the studio starts"); assert!( diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 2f8e4e8b..e54cdfb5 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -4,7 +4,7 @@ //! The scenes, sources, and compositor live in `moq-scene`; this binary is the //! window around them and the program clock that publishes the result. -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use anyhow::Context; @@ -12,6 +12,7 @@ use anyhow::Context; mod autosave; mod engine; mod health; +mod record; mod ui; #[derive(usage_rs::Cli, Clone)] @@ -27,12 +28,17 @@ pub struct Config { #[usage(long, default = "studio")] pub broadcast: String, - /// Copy and encode the program video only while someone watches it. This - /// saves the work while nobody watches, but leaves a late viewer nothing - /// from before they arrived. Audio is always encoded. + /// Copy and encode the program video only while someone watches or records + /// it. This saves the work while nobody does, but leaves a late viewer + /// nothing from before they arrived. Audio is always encoded. Without + /// `--connect`, the program and the sources are always on demand. #[usage(long)] pub on_demand: bool, + /// The directory recordings are saved in. Defaults to the scene file's. + #[usage(long, value_hint = usage_rs::ValueHint::DirPath)] + pub recordings: Option, + /// The MoQ client configuration. Without `--connect`, nothing is published /// and broadcast sources stay offline. #[usage(flatten)] @@ -72,6 +78,15 @@ fn main() -> anyhow::Result<()> { Some(url) => Some((config.client.clone().init(config.quic.clone())?, url)), None => None, }; + let recordings = match &config.recordings { + Some(dir) => dir.clone(), + None => config + .scenes + .parent() + .filter(|dir| !dir.as_os_str().is_empty()) + .unwrap_or(Path::new(".")) + .to_path_buf(), + }; let engine = runtime.block_on(engine::Engine::start( gpu.clone(), autosave, @@ -79,9 +94,14 @@ fn main() -> anyhow::Result<()> { connect, &config.broadcast, config.on_demand, + &recordings, ))?; let ran = run(engine.clone(), gpu, runtime.handle().clone()); + // A recording still running is finished, so it is a regular MP4. + if let Ok(recorder) = &engine.recorder { + runtime.block_on(recorder.finish()); + } // The edits since the last autosave, so quitting loses none. let saved = engine.save(); if let (Err(err), Err(_)) = (&saved, &ran) { diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs new file mode 100644 index 00000000..ac5e2aef --- /dev/null +++ b/rs/moq-studio/src/record.rs @@ -0,0 +1,756 @@ +//! Recording the program, and optionally each source, to disk beside any live +//! output. +//! +//! A recording reads the studio's broadcast the way a relay viewer does, so it +//! shares the live encoders and starts and stops without touching what the +//! relay serves. Each file is an MP4 that plays while it grows, so a crash +//! still leaves a playable file, and becomes a regular MP4 when the recording +//! stops. + +use std::collections::BTreeSet; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::task::{Poll, ready}; +use std::time::Duration; + +use anyhow::Context; +use kio::clock::Instant; +use moq_mux::catalog::{CatalogFormat, Stream}; +use moq_mux::container::mp4; +use tokio::io::{AsyncSeekExt, AsyncWriteExt}; +use tokio::sync::watch; + +/// How far a file may fall behind the broadcast before it skips ahead. Only a +/// disk stalled this long loses media; the studio never holds more for it. +const MAX_DELAY: Duration = Duration::from_secs(5); + +/// Records the program into a directory, one recording at a time. +pub struct Recorder { + source: moq_mux::Source, + dir: PathBuf, + /// Each file's name starts with this: the scene file's. + name: String, + runtime: tokio::runtime::Handle, + /// The recording in progress, or the last one. + session: Mutex>, +} + +struct Session { + started: Instant, + stop: Arc>, + done: watch::Receiver, + files: Files, +} + +/// Every file a recording has opened, in the order it opened them. +type Files = Arc>>>; + +/// One file a recording writes. +struct File { + path: PathBuf, + written: AtomicU64, + /// `None` while it is being written. + finished: Mutex>, +} + +/// How a file ended. +#[derive(Clone, Debug, PartialEq)] +pub enum Finished { + /// A regular MP4. + Saved, + /// Nothing reached it before the recording stopped, so it was removed. + Empty, + /// Writing stopped early. A file that holds anything still plays as far as + /// it got, as a fragmented MP4. + Failed(String), +} + +/// A recording, for the Output panel. +pub struct Status { + pub started: Instant, + /// Asked to stop, and finishing its files. + pub stopping: bool, + /// Every file has finished. + pub done: bool, + pub files: Vec, +} + +pub struct FileStatus { + pub path: PathBuf, + /// Bytes written so far. + pub written: u64, + /// `None` while it is being written. + pub finished: Option, +} + +impl Recorder { + /// Record `source`, the studio's broadcast, into `dir`, naming each file + /// after the `scenes` file. + /// + /// Call inside the tokio runtime the recordings write on. + pub fn new(source: moq_mux::Source, dir: &Path, scenes: &Path) -> anyhow::Result { + let name = scenes + .file_stem() + .context("the scene file has no name")? + .to_string_lossy() + .into_owned(); + Ok(Self { + source, + dir: std::path::absolute(dir)?, + name, + runtime: tokio::runtime::Handle::try_current().context("not inside a tokio runtime")?, + session: Mutex::new(None), + }) + } + + /// The directory recordings go to. + pub fn dir(&self) -> &Path { + &self.dir + } + + /// Start recording the program, and with `sources`, each source that + /// publishes video or audio into a file of its own, unless a recording is + /// still running. + pub fn start(&self, sources: bool) -> anyhow::Result<()> { + let mut session = self.session.lock().unwrap(); + anyhow::ensure!( + session.as_ref().is_none_or(|session| *session.done.borrow()), + "already recording" + ); + + // UTC, since the studio does not read the time zone; no colons, which + // Windows refuses in a file name. + let stamp = humantime::format_rfc3339_seconds(kio::clock::wall().into()) + .to_string() + .replace(':', "-"); + let stem = format!("{} {stamp}", self.name); + // Here rather than on the runtime, so a directory that cannot take it + // fails the click that started it. + let (path, file) = create(&self.dir, &stem)?; + let program = Arc::new(File::new(path)); + let files: Files = Arc::new(Mutex::new(vec![program.clone()])); + let stop = Arc::new(watch::channel(false).0); + let (finished, done) = watch::channel(false); + + let run = Run { + source: self.source.clone(), + dir: self.dir.clone(), + stem, + files: files.clone(), + stop: stop.clone(), + }; + let file = tokio::fs::File::from_std(file); + self.runtime.spawn(async move { + run.record(program, file, sources).await; + finished.send_replace(true); + }); + + *session = Some(Session { + started: kio::clock::now(), + stop, + done, + files, + }); + Ok(()) + } + + /// Ask the recording to stop. Its files finish in the background. + pub fn stop(&self) { + if let Some(session) = self.session.lock().unwrap().as_ref() { + session.stop.send_replace(true); + } + } + + /// Stop recording and wait until every file is finished. + pub async fn finish(&self) { + let mut done = { + let session = self.session.lock().unwrap(); + let Some(session) = session.as_ref() else { return }; + session.stop.send_replace(true); + session.done.clone() + }; + // Fails only once the recording's task is gone, having finished. + let _ = done.wait_for(|done| *done).await; + } + + /// The recording in progress, or the last one, or `None` before the first. + pub fn status(&self) -> Option { + let session = self.session.lock().unwrap(); + let session = session.as_ref()?; + let files = session.files.lock().unwrap(); + Some(Status { + started: session.started, + stopping: *session.stop.borrow(), + done: *session.done.borrow(), + files: files + .iter() + .map(|file| FileStatus { + path: file.path.clone(), + written: file.written.load(Ordering::Relaxed), + finished: file.finished.lock().unwrap().clone(), + }) + .collect(), + }) + } +} + +impl File { + fn new(path: PathBuf) -> Self { + Self { + path, + written: AtomicU64::new(0), + finished: Mutex::new(None), + } + } + + fn finish(&self, result: anyhow::Result) { + let finished = result.unwrap_or_else(|err| { + tracing::error!(path = %self.path.display(), error = format!("{err:#}"), "recording failed"); + Finished::Failed(format!("{err:#}")) + }); + *self.finished.lock().unwrap() = Some(finished); + } +} + +/// One recording, running on the runtime. +struct Run { + source: moq_mux::Source, + dir: PathBuf, + /// Each file's name starts with this. + stem: String, + files: Files, + stop: Arc>, +} + +impl Run { + /// Record the program into `file`, and with `sources` each source too, + /// until asked to stop or the program ends, and finish every file. + async fn record(self, program: Arc, file: tokio::fs::File, sources: bool) { + let mut writers = tokio::task::JoinSet::new(); + let source = self.source.clone(); + let asked = self.stop.subscribe(); + writers.spawn(async move { + program.finish(write(source, file, &program, asked).await); + None + }); + if sources && let Err(err) = self.sources(&mut writers).await { + tracing::error!(error = format!("{err:#}"), "cannot record the sources"); + self.stop.send_replace(true); + } + while writers.join_next().await.is_some() {} + } + + /// Give each source that publishes video or audio a file of its own while + /// it is listed, until asked to stop or the program's writer ends. + /// + /// A source removed while recording finishes its file; one added, or added + /// back, starts another. One whose file failed while it stays listed gets + /// no other. + async fn sources(&self, writers: &mut tokio::task::JoinSet>) -> anyhow::Result<()> { + let mut asked = self.stop.subscribe(); + let mut catalog = tokio::select! { + catalog = self.source.catalog::(CatalogFormat::Hang) => catalog?, + () = stopped(&mut asked) => return Ok(()), + }; + let mut listed = BTreeSet::new(); + let mut writing = BTreeSet::new(); + // Sources whose file ended while they were still listed last we looked. + let mut ended = BTreeSet::new(); + loop { + tokio::select! { + snapshot = catalog.next() => { + let Some(snapshot) = snapshot? else { return Ok(()) }; + listed = snapshot + .ext + .studio + .sources + .into_iter() + .filter(|(_, listing)| !listing.video.renditions.is_empty() || !listing.audio.renditions.is_empty()) + .map(|(name, _)| name) + .collect(); + ended.retain(|name| listed.contains(name)); + } + joined = writers.join_next() => match joined.context("no writer left")?? { + Some(name) => { + writing.remove(&name); + ended.insert(name); + } + // The program's writer ended, so the recording is over. + None => { + self.stop.send_replace(true); + return Ok(()); + } + }, + () = stopped(&mut asked) => return Ok(()), + } + let start: Vec = listed + .iter() + .filter(|name| !writing.contains(*name) && !ended.contains(*name)) + .cloned() + .collect(); + for name in start { + self.source_file(&name, writers); + writing.insert(name); + } + } + } + + /// Start recording the source `name` into a file of its own. + fn source_file(&self, name: &str, writers: &mut tokio::task::JoinSet>) { + // The scene file names sources, and a name may hold a path separator. + let safe: String = name + .chars() + .map(|c| match c { + c if c.is_alphanumeric() || " -_.()".contains(c) => c, + _ => '_', + }) + .collect(); + let stem = format!("{} {safe}", self.stem); + let (path, file) = match create(&self.dir, &stem) { + Ok(created) => created, + Err(err) => { + let entry = File::new(self.dir.join(format!("{stem}.mp4"))); + entry.finish(Err(err)); + self.files.lock().unwrap().push(Arc::new(entry)); + return; + } + }; + let entry = Arc::new(File::new(path)); + self.files.lock().unwrap().push(entry.clone()); + let (source, name) = (self.source.clone(), name.to_string()); + let asked = self.stop.subscribe(); + let file = tokio::fs::File::from_std(file); + writers.spawn(async move { + entry.finish(write_source(source, &name, file, &entry, asked).await); + Some(name) + }); + } +} + +/// Create a file named `stem` in `dir`, numbering it rather than replacing +/// one already there. +fn create(dir: &Path, stem: &str) -> anyhow::Result<(PathBuf, std::fs::File)> { + let mut number = 1; + loop { + let name = match number { + 1 => format!("{stem}.mp4"), + _ => format!("{stem} ({number}).mp4"), + }; + let path = dir.join(name); + match std::fs::File::create_new(&path) { + Ok(file) => return Ok((path, file)), + Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => number += 1, + Err(err) => return Err(err).with_context(|| format!("cannot create {}", path.display())), + } + } +} + +/// Record the program from `source` into `file` until it ends or `stop` is +/// set, then finish the file with everything it received. +async fn write( + source: moq_mux::Source, + file: tokio::fs::File, + entry: &File, + mut stop: watch::Receiver, +) -> anyhow::Result { + let catalog = tokio::select! { + catalog = source.catalog::<()>(CatalogFormat::Hang) => catalog.context("the program's catalog")?, + () = stopped(&mut stop) => return remove(file, entry).await, + }; + record(mp4::Export::new(source, catalog), file, entry, stop).await +} + +/// Record the source `name`'s own tracks from `source` into `file` until it +/// leaves the catalog or `stop` is set, then finish the file. +async fn write_source( + source: moq_mux::Source, + name: &str, + file: tokio::fs::File, + entry: &File, + mut stop: watch::Receiver, +) -> anyhow::Result { + let catalog = tokio::select! { + catalog = source.catalog::(CatalogFormat::Hang) => catalog.context("the studio's catalog")?, + () = stopped(&mut stop) => return remove(file, entry).await, + }; + let listing = Listing { + catalog, + name: name.to_string(), + listed: false, + }; + record(mp4::Export::new(source, listing), file, entry, stop).await +} + +/// Write `export` into `file` until it ends or `stop` is set, then finish it. +/// +/// A failed export still finishes what it recorded; a failed write leaves the +/// file as it is. +async fn record( + export: mp4::Export, + mut file: tokio::fs::File, + entry: &File, + mut stop: watch::Receiver, +) -> anyhow::Result { + let mut export = export.with_max_delay(MAX_DELAY); + let mut stopping = false; + let ended = loop { + let next = tokio::select! { + next = export.next() => next, + // What was read is still written. + () = stopped(&mut stop), if !stopping => { + export.stop(); + stopping = true; + continue; + } + }; + match next { + Ok(Some(data)) => { + file.write_all(&data).await.context("writing the recording")?; + entry.written.fetch_add(data.len() as u64, Ordering::Relaxed); + } + Ok(None) => break Ok(()), + Err(err) => break Err(err), + } + }; + + let Some(finish) = export.finish().context("finishing the recording")? else { + return remove(file, entry).await; + }; + // The moov where the media ends, then the mdat header over the reserved box, + // which points readers past the media at the moov, so the moov must be on + // disk first. + file.set_len(finish.moov_offset).await?; + file.seek(std::io::SeekFrom::Start(finish.moov_offset)).await?; + file.write_all(&finish.moov).await?; + file.sync_data().await?; + file.seek(std::io::SeekFrom::Start(finish.mdat_offset)).await?; + file.write_all(&finish.mdat).await?; + file.sync_all().await?; + entry + .written + .store(finish.moov_offset + finish.moov.len() as u64, Ordering::Relaxed); + + ended.context("reading the broadcast")?; + Ok(Finished::Saved) +} + +/// Wait until the recording is asked to stop, or its recorder is gone. +async fn stopped(stop: &mut watch::Receiver) { + let _ = stop.wait_for(|stop| *stop).await; +} + +/// Remove a file nothing was recorded into. +async fn remove(file: tokio::fs::File, entry: &File) -> anyhow::Result { + drop(file); + tokio::fs::remove_file(&entry.path).await?; + Ok(Finished::Empty) +} + +/// One source's own catalog, nested in the studio's, which ends once the +/// source leaves it. +struct Listing { + catalog: moq_mux::catalog::Consumer, + name: String, + /// Whether the source was listed yet. + listed: bool, +} + +impl Stream for Listing { + type Ext = (); + + fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { + loop { + let Some(studio) = ready!(self.catalog.poll_next(waiter))? else { + return Poll::Ready(Ok(None)); + }; + match studio.ext.studio.sources.get(&self.name) { + Some(listing) => { + self.listed = true; + return Poll::Ready(Ok(Some(listing.clone()))); + } + None if self.listed => return Poll::Ready(Ok(None)), + None => {} + } + } + } +} + +#[cfg(test)] +mod tests { + use bytes::Bytes; + use moq_mux::container::fmp4::mp4_atom::{self, DecodeMaybe}; + + use super::*; + + const RATE: u32 = 48_000; + /// One write's worth of audio, as the program's audio thread hands over. + const PERIOD: Duration = Duration::from_millis(10); + + /// A tone, published as an Opus track in `broadcast` and listed in `catalog`. + struct Tone { + audio: moq_audio::encode::Producer, + } + + impl Tone { + fn new( + broadcast: &mut moq_net::broadcast::Producer, + catalog: moq_mux::catalog::Producer, + ) -> Self { + let mut input = moq_audio::encode::Input::new(RATE, moq_audio::Layout::Stereo); + input.format = moq_audio::Format::F32; + let mut options = moq_audio::encode::Options::default(); + options.config = moq_audio::encode::Config::from_input(moq_audio::encode::Codec::Opus, &input); + let audio = moq_audio::encode::Producer::new(broadcast, catalog, input, &options).unwrap(); + Self { audio } + } + + /// Write the 10 ms of tone starting at `at`. + fn write(&mut self, at: Duration) { + let frames = (u128::from(RATE) * PERIOD.as_nanos() / 1_000_000_000) as usize; + let start = at.as_secs_f32(); + let data: Vec = (0..frames) + .map(|frame| (start + frame as f32 / RATE as f32) * 440.0 * std::f32::consts::TAU) + .flat_map(|phase| [phase.sin() * 0.5; 2]) + .flat_map(f32::to_le_bytes) + .collect(); + let timestamp = moq_net::Timestamp::try_from(at).unwrap(); + self.audio + .write(&moq_audio::Frame::new(Bytes::from(data), timestamp)) + .unwrap(); + } + } + + /// The studio's broadcast with only program audio, which needs no GPU, and + /// sources listed beside it as the studio lists them. + struct Studio { + origin: moq_net::origin::Producer, + publication: moq_net::origin::Publication, + catalog: moq_scene::publish::Catalog, + program: Tone, + /// Each listed source's tone, and the scratch broadcast its own catalog + /// writes into. + sources: Vec<(String, Tone, moq_net::broadcast::Producer, moq_mux::catalog::Producer)>, + /// How much audio was written. + played: Duration, + } + + impl Studio { + fn new() -> Self { + let origin = moq_tokio::origin::spawn(); + let publication = origin.create_broadcast("studio").unwrap(); + let mut broadcast = publication.broadcast().clone(); + let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); + let catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); + let program = Tone::new(&mut broadcast, catalog.clone()); + publication.announce(Default::default()).unwrap(); + Self { + origin, + publication, + catalog, + program, + sources: Vec::new(), + played: Duration::ZERO, + } + } + + fn recorder(&self, dir: &Path) -> Recorder { + let source = moq_mux::Source::new(self.origin.consume(), self.publication.broadcast().consume()); + Recorder::new(source, dir, Path::new("show.toml")).unwrap() + } + + /// List a source `name` publishing a tone of its own. + fn add(&mut self, name: &str) { + let mut scratch = moq_net::broadcast::Info::new().produce(); + let config = moq_mux::catalog::Config::default().with_clock(self.catalog.clock()); + let listing = moq_mux::catalog::Producer::new(&mut scratch, config).unwrap(); + let mut broadcast = self.publication.broadcast().clone(); + let tone = Tone::new(&mut broadcast, listing.clone()); + self.sources.push((name.to_string(), tone, scratch, listing)); + self.relist(); + } + + /// Stop listing the source `name`. + fn remove(&mut self, name: &str) { + self.sources.retain(|(source, ..)| source != name); + self.relist(); + } + + /// List what each source's own catalog holds, as the studio mirrors it. + fn relist(&mut self) { + let listings: Vec<_> = self + .sources + .iter() + .map(|(name, _, _, listing)| { + let mut listing = listing.snapshot(); + listing.clock = None; + (name.clone(), listing) + }) + .collect(); + self.catalog + .mutate(|root| root.ext.studio.sources = listings.into_iter().collect()) + .unwrap(); + } + + /// Play every tone for `duration`, as the program clock would. + async fn play(&mut self, duration: Duration) { + let end = self.played + duration; + while self.played < end { + self.program.write(self.played); + for (_, tone, ..) in &mut self.sources { + tone.write(self.played); + } + self.played += PERIOD; + kio::clock::sleep(PERIOD).await; + } + } + } + + /// Every track in `path`, by handler, and its duration. + fn tracks(path: &Path) -> Vec<(String, Duration)> { + let data = std::fs::read(path).unwrap(); + let mut cursor = std::io::Cursor::new(data.as_slice()); + let mut kinds = Vec::new(); + let mut tracks = Vec::new(); + while let Some(atom) = mp4_atom::Any::decode_maybe(&mut cursor).unwrap() { + kinds.push(atom.kind().to_string()); + if let mp4_atom::Any::Moov(moov) = atom { + for trak in moov.trak { + let mdhd = trak.mdia.mdhd; + let duration = Duration::from_secs_f64(mdhd.duration as f64 / f64::from(mdhd.timescale)); + tracks.push((trak.mdia.hdlr.handler.to_string(), duration)); + } + } + } + assert_eq!(kinds, ["ftyp", "mdat", "moov"], "a regular MP4"); + tracks + } + + /// Whether `duration` is about `expected`: within a group either way, for + /// where the recording joined and the fragment it finished. + fn about(duration: Duration, expected: Duration) -> bool { + duration.abs_diff(expected) <= Duration::from_millis(100) + } + + /// A recording started mid-program and stopped while it plays on is a + /// regular MP4 of what played in between. + #[tokio::test(start_paused = true)] + async fn records_a_regular_mp4_until_stopped() { + let dir = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + let recorder = studio.recorder(dir.path()); + assert!(recorder.status().is_none()); + + studio.play(Duration::from_millis(500)).await; + recorder.start(false).unwrap(); + assert!(recorder.start(false).is_err(), "one recording at a time"); + studio.play(Duration::from_secs(1)).await; + recorder.stop(); + let ((), ()) = tokio::join!(recorder.finish(), studio.play(Duration::from_millis(500))); + + let status = recorder.status().unwrap(); + assert!(status.stopping && status.done); + let [file] = status.files.as_slice() else { + panic!("{} files", status.files.len()); + }; + assert_eq!(file.finished, Some(Finished::Saved)); + let name = file.path.file_name().unwrap().to_string_lossy(); + assert!(name.starts_with("show ") && name.ends_with("Z.mp4"), "{name}"); + assert_eq!(file.written, std::fs::metadata(&file.path).unwrap().len()); + let [(handler, duration)] = tracks(&file.path).try_into().unwrap(); + assert_eq!(handler, "soun"); + assert!(about(duration, Duration::from_secs(1)), "{duration:?}"); + + // The next recording starts fresh. + recorder.start(false).unwrap(); + assert_eq!(recorder.status().unwrap().files.len(), 1); + recorder.finish().await; + } + + /// Each source with media of its own records into its own file while it is + /// listed, beside the program; a still has nothing to record. + #[tokio::test(start_paused = true)] + async fn sources_record_into_their_own_files() { + let dir = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + studio.add("mic/1"); + studio + .catalog + .mutate(|root| { + let mut still = hang::Catalog::default(); + still.binary.tracks.insert( + "logo.png".into(), + hang::catalog::BinaryConfig::new(hang::catalog::Mode::Snapshot), + ); + root.ext.studio.sources.insert("logo".into(), still); + }) + .unwrap(); + let recorder = studio.recorder(dir.path()); + + recorder.start(true).unwrap(); + studio.play(Duration::from_secs(1)).await; + studio.add("guitar"); + studio.play(Duration::from_secs(1)).await; + // Its file finishes while the rest record on. + studio.remove("mic/1"); + studio.play(Duration::from_secs(1)).await; + recorder.finish().await; + + let status = recorder.status().unwrap(); + let files: Vec<_> = status + .files + .iter() + .map(|file| { + assert_eq!(file.finished, Some(Finished::Saved), "{}", file.path.display()); + let name = file.path.file_name().unwrap().to_string_lossy().into_owned(); + let [(handler, duration)] = tracks(&file.path).try_into().unwrap(); + assert_eq!(handler, "soun"); + (name.rsplit_once("Z").unwrap().1.to_string(), duration) + }) + .collect(); + let names: Vec<_> = files.iter().map(|(name, _)| name.as_str()).collect(); + assert_eq!(names, [".mp4", " mic_1.mp4", " guitar.mp4"]); + for ((_, duration), expected) in files.iter().zip([3, 2, 2]) { + assert!(about(*duration, Duration::from_secs(expected)), "{files:?}"); + } + } + + /// A recording stopped before any media arrived leaves no file behind. + #[tokio::test(start_paused = true)] + async fn an_empty_recording_is_removed() { + let dir = tempfile::tempdir().unwrap(); + let studio = Studio::new(); + let recorder = studio.recorder(dir.path()); + recorder.start(true).unwrap(); + recorder.finish().await; + + let status = recorder.status().unwrap(); + assert_eq!(status.files.len(), 1); + assert_eq!(status.files[0].finished, Some(Finished::Empty)); + assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0); + } + + /// A name already taken, such as by a recording started in the same + /// second, is numbered rather than replaced. + #[test] + fn a_taken_name_is_numbered() { + let dir = tempfile::tempdir().unwrap(); + let (first, _) = create(dir.path(), "show").unwrap(); + let (second, _) = create(dir.path(), "show").unwrap(); + let (third, _) = create(dir.path(), "show").unwrap(); + assert_eq!(first, dir.path().join("show.mp4")); + assert_eq!(second, dir.path().join("show (2).mp4")); + assert_eq!(third, dir.path().join("show (3).mp4")); + } + + /// Starting into a directory that does not exist fails at once. + #[tokio::test(start_paused = true)] + async fn a_missing_directory_fails_to_start() { + let dir = tempfile::tempdir().unwrap(); + let studio = Studio::new(); + let recorder = studio.recorder(&dir.path().join("missing")); + let err = recorder.start(false).unwrap_err(); + assert!(format!("{err:#}").contains("cannot create"), "{err:#}"); + assert!(recorder.status().is_none()); + } +} diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 7233fcb2..8c3fc87b 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -12,6 +12,7 @@ use moq_scene::{Clip, Collection, Draft, Filter, Item, Layer, Rect, Scene, Sourc use crate::engine::{Engine, State}; use crate::health::{self, Verdict}; +use crate::record; /// The panels, each a tab the user can move, split, or stack. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] @@ -102,6 +103,7 @@ impl Studio { new_item: None, add: AddSource::default(), devices: Devices::default(), + record_sources: false, message: (dropped > 0).then(|| Err(dropped_message(dropped))), }, }) @@ -139,6 +141,8 @@ struct Panels { new_item: Option, add: AddSource, devices: Devices, + /// Whether the next recording gives each source a file of its own too. + record_sources: bool, /// The outcome of the last action, shown in the status bar. message: Option>, } @@ -1019,10 +1023,64 @@ impl Panels { } } + /// Start and stop recording, and see what the recording wrote. + fn record(&mut self, ui: &mut Ui) { + let engine = self.engine.clone(); + let recorder = match &engine.recorder { + Ok(recorder) => recorder, + Err(reason) => { + ui.colored_label(ui.visuals().error_fg_color, format!("Cannot record: {reason}")); + return; + } + }; + let status = recorder.status(); + ui.horizontal(|ui| match status.as_ref().filter(|status| !status.done) { + Some(status) => { + if ui + .add_enabled(!status.stopping, egui::Button::new("Stop recording")) + .clicked() + { + recorder.stop(); + } + if status.stopping { + ui.label("finishing"); + } else { + let elapsed = kio::clock::now().saturating_duration_since(status.started).as_secs(); + let elapsed = format!("{}:{:02}:{:02}", elapsed / 3600, elapsed / 60 % 60, elapsed % 60); + ui.colored_label(ui.visuals().error_fg_color, format!("recording {elapsed}")); + } + } + None => { + if ui.button("Start recording").clicked() { + let result = recorder.start(self.record_sources); + self.check(result); + } + ui.checkbox(&mut self.record_sources, "each source too"); + ui.weak(format!("into {}", recorder.dir().display())); + } + }); + for file in status.iter().flat_map(|status| &status.files) { + let name = file.path.file_name().unwrap_or_default().to_string_lossy(); + let size = format!("{:.1} MB", file.written as f64 / 1e6); + match &file.finished { + None => ui.label(format!("{name}: {size}")), + Some(record::Finished::Saved) => ui.label(format!("{name}: saved, {size}")), + Some(record::Finished::Empty) => ui.weak(format!("{name}: nothing recorded, removed")), + Some(record::Finished::Failed(reason)) => { + ui.colored_label(ui.visuals().error_fg_color, format!("{name}: {reason}")) + } + }; + } + } + fn output(&mut self, ui: &mut Ui) { + self.record(ui); + ui.separator(); + let stats = &self.engine.stats; let count = |value: &std::sync::atomic::AtomicU64| value.load(Ordering::Relaxed); let millis = |value: &std::sync::atomic::AtomicU64| format!("{:.1} ms", count(value) as f64 / 1000.0); + let watched = stats.watched.load(Ordering::Relaxed); egui::Grid::new("output").show(ui, |ui| { match &self.engine.output { @@ -1031,10 +1089,10 @@ impl Panels { ui.label(&output.path); ui.end_row(); ui.label("Encoding"); - ui.label(match (output.on_demand, stats.watched.load(Ordering::Relaxed)) { + ui.label(match (output.on_demand, watched) { (false, _) => "always", - (true, true) => "while watched; watched now", - (true, false) => "while watched; idle now", + (true, true) => "while watched or recorded; wanted now", + (true, false) => "while watched or recorded; idle now", }); ui.end_row(); } @@ -1042,6 +1100,15 @@ impl Panels { ui.label("Relay"); ui.label("none; start with --connect to publish"); ui.end_row(); + if self.engine.recorder.is_ok() { + ui.label("Encoding"); + ui.label(if watched { + "while recorded; recording now" + } else { + "while recorded; idle now" + }); + ui.end_row(); + } } } ui.label("Drawn"); From b5fbe540885b84b311f4d98ca96bf4384a8513d3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 22:36:30 -0800 Subject: [PATCH 03/17] feat(studio): finish recording, delete the quest Co-Authored-By: Claude Opus 5.5 --- quest/m0/README.md | 1 - quest/m0/recording.md | 14 -------------- quest/m0/replay.md | 3 ++- rs/moq-studio/src/engine.rs | 6 ++++++ rs/moq-studio/src/record.rs | 32 +++++++++++++++++++++++--------- 5 files changed, 31 insertions(+), 25 deletions(-) delete mode 100644 quest/m0/recording.md diff --git a/quest/m0/README.md b/quest/m0/README.md index e39b046c..26fbd72b 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -98,7 +98,6 @@ repository. - [Guests](/quest/m0/guests.md) - remote guests and phone cameras join through a moq-room link or QR code and appear as sources - [Audio mixes](/quest/m0/mixes.md) - per-output mixes, meters, monitoring, and sync offsets - [Audio filters](/quest/m0/audio-filters.md) - per-source noise suppression, gate, compressor, limiter, and plugins, configured from the studio -- [Recording](/quest/m0/recording.md) - each canvas's program, and optionally each source, recorded to disk alongside live outputs - [Browser sources](/quest/m0/browser-source.md) - web overlays rendered by out-of-process CEF into the compositor through shared GPU textures - [Web studio](/quest/m0/web.md) - the full studio in wasm on WebGPU and WebCodecs, served from this repository - [Virtual camera](/quest/m0/vcam/README.md) - a canvas's program exposed as a system camera on Linux, Windows, and macOS diff --git a/quest/m0/recording.md b/quest/m0/recording.md deleted file mode 100644 index 5ccb1138..00000000 --- a/quest/m0/recording.md +++ /dev/null @@ -1,14 +0,0 @@ -# [M] Studio recording - -## Goal - -The studio records each canvas's program to disk, and optionally each source -as its own isolated track, independent of whether it is live. Recording -starts and stops on its own without touching live outputs. - -## Plan - -Write fMP4 through moq-mux. A recording shares a live destination's encoder -only when both have the same input (the same canvas or source, and the same -audio mix) and identical encode settings. An isolated source track never -shares the program's encoder. diff --git a/quest/m0/replay.md b/quest/m0/replay.md index 69615432..663441f4 100644 --- a/quest/m0/replay.md +++ b/quest/m0/replay.md @@ -27,8 +27,9 @@ case where every viewer composes sources themselves and nothing outside the studio subscribes to the program. Replay published to viewers is a new source in the scene, so it composes like any other. +A clip saves through the studio's recorder (`rs/moq-studio/src/record.rs`), +which already writes a broadcast, or one source's listing, to a crash-safe MP4. ## Related - [Archive DVR](https://github.com/moq-dev/moq/blob/main/quest/m2/archive-dvr.md) - viewer scrubbing beyond the relay cache -- [Recording](/quest/m0/recording.md) - clips use the same file writer diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 0f917e5f..338b072e 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -1112,5 +1112,11 @@ mod tests { }) .await .expect("health grades the program's tracks"); + + // A recording beside the live program reads the encoder viewers do. + let encoded = engine.stats.encoded.load(Ordering::Relaxed); + let files = record(&engine, encoded + 60, true).await; + assert_eq!(files, [["soun", "vide"]]); + assert!(engine.stats.error.lock().unwrap().is_none()); } } diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index ac5e2aef..a00b1121 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -21,9 +21,11 @@ use moq_mux::container::mp4; use tokio::io::{AsyncSeekExt, AsyncWriteExt}; use tokio::sync::watch; -/// How far a file may fall behind the broadcast before it skips ahead. Only a -/// disk stalled this long loses media; the studio never holds more for it. -const MAX_DELAY: Duration = Duration::from_secs(5); +/// How far a file may fall behind the broadcast before it skips ahead: only a +/// disk stalled this long loses media, and the studio never holds more for it. +/// A recording also starts with what the broadcast still holds from this far +/// before it started, since moq-mux reads from the oldest group within it. +pub const MAX_DELAY: Duration = Duration::from_secs(2); /// Records the program into a directory, one recording at a time. pub struct Recorder { @@ -572,6 +574,10 @@ mod tests { /// Stop listing the source `name`. fn remove(&mut self, name: &str) { + // A removed source ends its track cleanly, as the studio's do. + for (_, tone, ..) in self.sources.iter_mut().filter(|(source, ..)| source == name) { + tone.audio.finish().unwrap(); + } self.sources.retain(|(source, ..)| source != name); self.relist(); } @@ -633,7 +639,8 @@ mod tests { } /// A recording started mid-program and stopped while it plays on is a - /// regular MP4 of what played in between. + /// regular MP4 of what played in between, from [`MAX_DELAY`] before it + /// started. #[tokio::test(start_paused = true)] async fn records_a_regular_mp4_until_stopped() { let dir = tempfile::tempdir().unwrap(); @@ -641,7 +648,7 @@ mod tests { let recorder = studio.recorder(dir.path()); assert!(recorder.status().is_none()); - studio.play(Duration::from_millis(500)).await; + studio.play(Duration::from_secs(5)).await; recorder.start(false).unwrap(); assert!(recorder.start(false).is_err(), "one recording at a time"); studio.play(Duration::from_secs(1)).await; @@ -659,7 +666,7 @@ mod tests { assert_eq!(file.written, std::fs::metadata(&file.path).unwrap().len()); let [(handler, duration)] = tracks(&file.path).try_into().unwrap(); assert_eq!(handler, "soun"); - assert!(about(duration, Duration::from_secs(1)), "{duration:?}"); + assert!(about(duration, MAX_DELAY + Duration::from_secs(1)), "{duration:?}"); // The next recording starts fresh. recorder.start(false).unwrap(); @@ -715,13 +722,20 @@ mod tests { } } - /// A recording stopped before any media arrived leaves no file behind. + /// A recording of a broadcast with nothing to describe yet, stopped, leaves + /// no file behind. #[tokio::test(start_paused = true)] async fn an_empty_recording_is_removed() { let dir = tempfile::tempdir().unwrap(); - let studio = Studio::new(); - let recorder = studio.recorder(dir.path()); + let origin = moq_tokio::origin::spawn(); + let publication = origin.create_broadcast("studio").unwrap(); + let mut broadcast = publication.broadcast().clone(); + let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); + let _catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); + let source = moq_mux::Source::new(origin.consume(), publication.broadcast().consume()); + let recorder = Recorder::new(source, dir.path(), Path::new("show.toml")).unwrap(); recorder.start(true).unwrap(); + kio::clock::sleep(Duration::from_secs(1)).await; recorder.finish().await; let status = recorder.status().unwrap(); From 2d1ed8f964aed445116b2b1ac4d805b2e8f8f1b4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 06:29:47 -0800 Subject: [PATCH 04/17] fix(studio): record a source again once it is listed again Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index a00b1121..fbf3968f 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -274,9 +274,13 @@ impl Run { ended.retain(|name| listed.contains(name)); } joined = writers.join_next() => match joined.context("no writer left")?? { + // Held back only while still listed: one already gone may be + // listed again later, and is then recorded afresh. Some(name) => { writing.remove(&name); - ended.insert(name); + if listed.contains(&name) { + ended.insert(name); + } } // The program's writer ended, so the recording is over. None => { @@ -675,7 +679,8 @@ mod tests { } /// Each source with media of its own records into its own file while it is - /// listed, beside the program; a still has nothing to record. + /// listed, beside the program, and into another when listed again; a still + /// has nothing to record. #[tokio::test(start_paused = true)] async fn sources_record_into_their_own_files() { let dir = tempfile::tempdir().unwrap(); @@ -701,6 +706,9 @@ mod tests { // Its file finishes while the rest record on. studio.remove("mic/1"); studio.play(Duration::from_secs(1)).await; + // Added back, it records into another file. + studio.add("mic/1"); + studio.play(Duration::from_secs(1)).await; recorder.finish().await; let status = recorder.status().unwrap(); @@ -716,8 +724,8 @@ mod tests { }) .collect(); let names: Vec<_> = files.iter().map(|(name, _)| name.as_str()).collect(); - assert_eq!(names, [".mp4", " mic_1.mp4", " guitar.mp4"]); - for ((_, duration), expected) in files.iter().zip([3, 2, 2]) { + assert_eq!(names, [".mp4", " mic_1.mp4", " guitar.mp4", " mic_1 (2).mp4"]); + for ((_, duration), expected) in files.iter().zip([4, 2, 3, 1]) { assert!(about(*duration, Duration::from_secs(expected)), "{files:?}"); } } From af75f5edb0491c6cf0482f5aed37b30fd558997d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 06:35:51 -0800 Subject: [PATCH 05/17] fix(studio): judge a source writer by its own listing's lifetime Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index fbf3968f..42195bf3 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -7,7 +7,7 @@ //! still leaves a playable file, and becomes a regular MP4 when the recording //! stops. -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; @@ -256,8 +256,11 @@ impl Run { () = stopped(&mut asked) => return Ok(()), }; let mut listed = BTreeSet::new(); - let mut writing = BTreeSet::new(); - // Sources whose file ended while they were still listed last we looked. + // Each source with a file being written, and whether its listing has + // gone since that file started. + let mut writing = BTreeMap::new(); + // Sources whose file ended while their listing stayed, such as on a + // failure, held back until they leave the catalog. let mut ended = BTreeSet::new(); loop { tokio::select! { @@ -272,13 +275,15 @@ impl Run { .map(|(name, _)| name) .collect(); ended.retain(|name| listed.contains(name)); + for (name, gone) in &mut writing { + *gone |= !listed.contains(name); + } } joined = writers.join_next() => match joined.context("no writer left")?? { - // Held back only while still listed: one already gone may be - // listed again later, and is then recorded afresh. + // A listing that went away and came back is recorded afresh. Some(name) => { - writing.remove(&name); - if listed.contains(&name) { + let gone = writing.remove(&name).unwrap_or_default(); + if !gone && listed.contains(&name) { ended.insert(name); } } @@ -292,12 +297,12 @@ impl Run { } let start: Vec = listed .iter() - .filter(|name| !writing.contains(*name) && !ended.contains(*name)) + .filter(|name| !writing.contains_key(*name) && !ended.contains(*name)) .cloned() .collect(); for name in start { self.source_file(&name, writers); - writing.insert(name); + writing.insert(name, false); } } } From 1b79b84abac15d8db1383b7bb4a40f0bcc94f5e7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 06:39:38 -0800 Subject: [PATCH 06/17] fix(studio): name the writer table's type Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 42195bf3..61011e5d 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -258,7 +258,7 @@ impl Run { let mut listed = BTreeSet::new(); // Each source with a file being written, and whether its listing has // gone since that file started. - let mut writing = BTreeMap::new(); + let mut writing: BTreeMap = BTreeMap::new(); // Sources whose file ended while their listing stayed, such as on a // failure, held back until they leave the catalog. let mut ended = BTreeSet::new(); From a40f14e712631fc70767483fa405f33d7f5a1d57 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 17:43:20 -0800 Subject: [PATCH 07/17] docs(readme): broadcast sources and guests get no recording of their own Co-Authored-By: Claude Opus 5.5 --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 0bad8710..aee4b494 100644 --- a/README.md +++ b/README.md @@ -397,7 +397,8 @@ otherwise. or not. With **each source too**, every camera, display, and window that publishes a track of its own records into a file of its own beside it, ` .mp4`, holding the capture before -any filter. A source added while recording starts its file then, and one +any filter. Broadcast sources, guests included, publish no track of the +studio's, so they get no file of their own. A source added while recording starts its file then, and one removed finishes its file then. A recording reads the broadcast the way a viewer does, so it shares the live encoders and starts and stops without touching what the relay serves. Without `--connect`, or with `--on-demand`, From 3467cef20e201a6a755595b901ebecdd1cde8fbb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 17:59:10 -0800 Subject: [PATCH 08/17] fix(studio): a source whose track is replaced goes on in a new file A capture that changes size or color space moves to a new track. The source's recording took it as a second video track in the same MP4, which players skip, so the file stopped showing the capture at the resize. The file now ends there, keeping the old track to its end, and the source goes on in a numbered file of its own. Co-Authored-By: Claude Opus 5.5 --- README.md | 6 +- rs/moq-studio/src/record.rs | 153 ++++++++++++++++++++++++++++++------ 2 files changed, 132 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index aee4b494..6b1a6239 100644 --- a/README.md +++ b/README.md @@ -398,8 +398,10 @@ or not. With **each source too**, every camera, display, and window that publishes a track of its own records into a file of its own beside it, ` .mp4`, holding the capture before any filter. Broadcast sources, guests included, publish no track of the -studio's, so they get no file of their own. A source added while recording starts its file then, and one -removed finishes its file then. A recording reads the broadcast the way a +studio's, so they get no file of their own. A source added while recording +starts its file then, and one removed finishes its file then. A capture that +changes size or color space goes on in a new file, numbered after the first, +since players show only one video track. A recording reads the broadcast the way a viewer does, so it shares the live encoders and starts and stops without touching what the relay serves. Without `--connect`, or with `--on-demand`, the program encodes only while something records or watches it. diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 1da17237..e1819e48 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -9,7 +9,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::task::{Poll, ready}; use std::time::Duration; @@ -225,6 +225,17 @@ struct Run { stop: Arc>, } +/// A recording's file writers, each ending with the source whose file it +/// wrote, or `None` for the program's. +type Writers = tokio::task::JoinSet>; + +/// A source's file that ended. +struct Ended { + name: String, + /// The source's tracks were replaced, so it goes on in another file. + replaced: bool, +} + impl Run { /// Record the program into `file`, and with `sources` each source too, /// until asked to stop or the program ends, and finish every file. @@ -247,9 +258,9 @@ impl Run { /// it is listed, until asked to stop or the program's writer ends. /// /// A source removed while recording finishes its file; one added, or added - /// back, starts another. One whose file failed while it stays listed gets - /// no other. - async fn sources(&self, writers: &mut tokio::task::JoinSet>) -> anyhow::Result<()> { + /// back, starts another, as does one whose tracks were replaced. One whose + /// file failed while it stays listed gets no other. + async fn sources(&self, writers: &mut Writers) -> anyhow::Result<()> { let mut asked = self.stop.subscribe(); let mut catalog = tokio::select! { catalog = self.source.catalog::(CatalogFormat::Hang) => catalog?, @@ -280,10 +291,11 @@ impl Run { } } joined = writers.join_next() => match joined.context("no writer left")?? { - // A listing that went away and came back is recorded afresh. - Some(name) => { + // A listing that went away and came back, or whose tracks were + // replaced, is recorded afresh. + Some(Ended { name, replaced }) => { let gone = writing.remove(&name).unwrap_or_default(); - if !gone && listed.contains(&name) { + if !gone && !replaced && listed.contains(&name) { ended.insert(name); } } @@ -312,7 +324,7 @@ impl Run { /// Start recording the source `name` into a file of its own, returning /// whether it started rather than failing to create the file. - fn source_file(&self, name: &str, writers: &mut tokio::task::JoinSet>) -> bool { + fn source_file(&self, name: &str, writers: &mut Writers) -> bool { // The scene file names sources, and a name may hold a path separator. let safe: String = name .chars() @@ -337,8 +349,9 @@ impl Run { let asked = self.stop.subscribe(); let file = tokio::fs::File::from_std(file); writers.spawn(async move { - entry.finish(write_source(source, &name, file, &entry, asked).await); - Some(name) + let (finished, replaced) = write_source(source, &name, file, &entry, asked).await; + entry.finish(finished); + Some(Ended { name, replaced }) }); true } @@ -378,24 +391,32 @@ async fn write( } /// Record the source `name`'s own tracks from `source` into `file` until it -/// leaves the catalog or `stop` is set, then finish the file. +/// leaves the catalog, its tracks are replaced, or `stop` is set, then finish +/// the file. Also returns whether its tracks were replaced. async fn write_source( source: moq_mux::Source, name: &str, file: tokio::fs::File, entry: &File, mut stop: watch::Receiver, -) -> anyhow::Result { +) -> (anyhow::Result, bool) { let catalog = tokio::select! { - catalog = source.catalog::(CatalogFormat::Hang) => catalog.context("the studio's catalog")?, - () = stopped(&mut stop) => return remove(file, entry).await, + catalog = source.catalog::(CatalogFormat::Hang) => catalog.context("the studio's catalog"), + () = stopped(&mut stop) => return (remove(file, entry).await, false), }; + let catalog = match catalog { + Ok(catalog) => catalog, + Err(err) => return (Err(err), false), + }; + let replaced = Arc::new(AtomicBool::new(false)); let listing = Listing { catalog, name: name.to_string(), - listed: false, + last: None, + replaced: replaced.clone(), }; - record(mp4::Export::new(source, listing), file, entry, stop).await + let finished = record(mp4::Export::new(source, listing), file, entry, stop).await; + (finished, replaced.load(Ordering::Relaxed)) } /// Write `export` into `file` until it ends or `stop` is set, then finish it. @@ -464,34 +485,67 @@ async fn remove(file: tokio::fs::File, entry: &File) -> anyhow::Result } /// One source's own catalog, nested in the studio's, which ends once the -/// source leaves it. +/// source leaves it or its tracks are replaced. +/// +/// A capture that changes size or color space moves to a new track, which an +/// MP4 would hold as a second track that players skip, so the new one goes +/// into another file instead. struct Listing { catalog: moq_mux::catalog::Consumer, name: String, - /// Whether the source was listed yet. - listed: bool, + /// The source's catalog as last recorded, once it was listed. + last: Option, + /// Set as the listing ends because its tracks were replaced. + replaced: Arc, } impl Stream for Listing { type Ext = (); fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>> { + if self.replaced.load(Ordering::Relaxed) { + return Poll::Ready(Ok(None)); + } loop { let Some(studio) = ready!(self.catalog.poll_next(waiter))? else { return Poll::Ready(Ok(None)); }; - match studio.ext.studio.sources.get(&self.name) { - Some(listing) => { - self.listed = true; - return Poll::Ready(Ok(Some(listing.clone()))); + let Some(listing) = studio.ext.studio.sources.get(&self.name) else { + if self.last.is_some() { + return Poll::Ready(Ok(None)); + } + continue; + }; + if let Some(last) = &self.last { + let video = replaced(&last.video.renditions, &listing.video.renditions); + let audio = replaced(&last.audio.renditions, &listing.audio.renditions); + if video || audio { + // The replaced tracks stay to the end their publisher gives + // them, and the rest go on in the next file. + let mut tail = last.clone(); + if !video { + tail.video.renditions.clear(); + } + if !audio { + tail.audio.renditions.clear(); + } + self.replaced.store(true, Ordering::Relaxed); + return Poll::Ready(Ok(Some(tail))); } - None if self.listed => return Poll::Ready(Ok(None)), - None => {} } + self.last = Some(listing.clone()); + return Poll::Ready(Ok(Some(listing.clone()))); } } } +/// Whether `now` replaced the renditions of a kind the file already records. +/// A kind that first appears joins the file instead, and a rendition's config +/// changing under the same name stays on its track. +fn replaced(last: &BTreeMap, now: &BTreeMap) -> bool { + !last.is_empty() && !last.keys().eq(now.keys()) +} + #[cfg(test)] mod tests { use bytes::Bytes; @@ -596,6 +650,21 @@ mod tests { self.relist(); } + /// Move the source `name` onto a new track, as a capture that changes + /// size is. + fn replace(&mut self, name: &str) { + let mut broadcast = self.publication.broadcast().clone(); + let (_, tone, _, listing) = self.sources.iter_mut().find(|(source, ..)| source == name).unwrap(); + let old = listing.snapshot().audio.renditions; + let mut old_tone = std::mem::replace(tone, Tone::new(&mut broadcast, listing.clone())); + old_tone.audio.finish().unwrap(); + drop(old_tone); + listing + .mutate(|catalog| catalog.audio.renditions.retain(|track, _| !old.contains_key(track))) + .unwrap(); + self.relist(); + } + /// List what each source's own catalog holds, as the studio mirrors it. fn relist(&mut self) { let listings: Vec<_> = self @@ -740,6 +809,40 @@ mod tests { } } + /// A source moved onto a new track, as a capture that changes size is, goes + /// on in another file, so neither holds a track that players skip. + #[tokio::test(start_paused = true)] + async fn a_replaced_track_goes_on_in_another_file() { + let dir = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + studio.add("camera"); + let recorder = studio.recorder(dir.path()); + + recorder.start(true).unwrap(); + studio.play(Duration::from_secs(1)).await; + studio.replace("camera"); + studio.play(Duration::from_secs(2)).await; + recorder.finish().await; + + let status = recorder.status().unwrap(); + let files: Vec<_> = status + .files + .iter() + .map(|file| { + assert_eq!(file.finished, Some(Finished::Saved), "{}", file.path.display()); + let name = file.path.file_name().unwrap().to_string_lossy().into_owned(); + let [(handler, duration)] = tracks(&file.path).try_into().unwrap(); + assert_eq!(handler, "soun"); + (name.rsplit_once("Z").unwrap().1.to_string(), duration) + }) + .collect(); + let names: Vec<_> = files.iter().map(|(name, _)| name.as_str()).collect(); + assert_eq!(names, [".mp4", " camera.mp4", " camera (2).mp4"]); + for ((_, duration), expected) in files.iter().zip([3, 1, 2]) { + assert!(about(*duration, Duration::from_secs(expected)), "{files:?}"); + } + } + /// A source whose file cannot be created fails without stopping the rest, /// and once removed and added back, records again. #[cfg(unix)] From fb4b778ca4419b0c23ac0d7e452ed16414431f07 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 19:27:36 -0800 Subject: [PATCH 09/17] fix(studio): a recording that fails before any media says why An export that failed before writing anything had its error dropped with the empty file, so the Output panel showed it as nothing recorded. The README also says a file that goes on in the next repeats up to 2 s. Co-Authored-By: Claude Opus 5.5 --- README.md | 5 ++++- rs/moq-studio/src/record.rs | 37 ++++++++++++++++++++++++++++++++++++- 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 6b61b2ce..355a0cc3 100644 --- a/README.md +++ b/README.md @@ -410,7 +410,10 @@ studio's, so they get no file of their own. A source added while recording starts its file then, and one removed finishes its file then. A new canvas size or frame rate, or a capture that changes size or color space, moves the video to a new track, so the program or that source goes on in a new file, -numbered after the first, since players show only one video track. A +numbered after the first, since players show only one video track. A file +starts with up to 2 s from before it opened, so a recording starts a little +early, and a track that goes on into the next file, such as the program +audio, repeats up to 2 s of the last. A recording reads the broadcast the way a viewer does, so it shares the live encoders and starts and stops without touching what the relay serves. Without `--connect`, or with `--on-demand`, the program encodes only while something diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index bcd331df..add53566 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -462,7 +462,10 @@ async fn record( }; let Some(finish) = export.finish().context("finishing the recording")? else { - return remove(file, entry).await; + // Removed either way, but a broadcast that failed says why. + let removed = remove(file, entry).await; + ended.context("reading the broadcast")?; + return removed; }; // The moov where the media ends, then the mdat header over the reserved box, // which points readers past the media at the moov, so the moov must be on @@ -978,6 +981,38 @@ mod tests { assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0); } + /// A broadcast that fails before anything reached the file still says why, + /// though the empty file is removed. + #[tokio::test(start_paused = true)] + async fn a_failure_before_any_media_says_why() { + let dir = tempfile::tempdir().unwrap(); + let origin = moq_tokio::origin::spawn(); + let publication = origin.create_broadcast("studio").unwrap(); + let mut broadcast = publication.broadcast().clone(); + let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); + let catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); + publication.announce(Default::default()).unwrap(); + let source = moq_mux::Source::new(origin.consume(), publication.broadcast().consume()); + let recorder = Recorder::new(source, dir.path(), Path::new("show.toml")).unwrap(); + recorder.start(false).unwrap(); + kio::clock::sleep(Duration::from_millis(100)).await; + catalog.abort(moq_net::StreamError::Internal.into()); + kio::clock::timeout(Duration::from_secs(5), recorder.finish()) + .await + .expect("the recording ends with its program"); + + let status = recorder.status().unwrap(); + let [file] = status.files.as_slice() else { + panic!("{} files", status.files.len()); + }; + assert!( + matches!(&file.finished, Some(Finished::Failed(reason)) if reason.contains("reading the broadcast")), + "{:?}", + file.finished + ); + assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0); + } + /// A name already taken, such as by a recording started in the same /// second, is numbered rather than replaced. #[test] From 1e4115933c78ae638b5dd862ccaf221f8f8fabf6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 20:01:00 -0800 Subject: [PATCH 10/17] fix(studio): a long scene or source name is cut so its recording file fits A source's sanitized name had no length bound, so a long one made the file name exceed what file systems take and that source got no file. The scene stem and the source's name are cut, between characters, so each name fits in 255 bytes with the largest number and the extension. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 78 ++++++++++++++++++++++++++++++------- 1 file changed, 64 insertions(+), 14 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index add53566..f9b0c9f9 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -126,7 +126,7 @@ impl Recorder { let stamp = humantime::format_rfc3339_seconds(kio::clock::wall().into()) .to_string() .replace(':', "-"); - let stem = format!("{} {stamp}", self.name); + let stem = program_stem(&self.name, &stamp); // Here rather than on the runtime, so a directory that cannot take it // fails the click that started it. let (path, file) = create(&self.dir, &stem)?; @@ -337,17 +337,7 @@ impl Run { fn file(&self, source: Option<&str>, writers: &mut Writers) -> bool { let stem = match source { None => self.stem.clone(), - Some(name) => { - // The scene file names sources, and a name may hold a path separator. - let safe: String = name - .chars() - .map(|c| match c { - c if c.is_alphanumeric() || " -_.()".contains(c) => c, - _ => '_', - }) - .collect(); - format!("{} {safe}", self.stem) - } + Some(name) => source_stem(&self.stem, name), }; let (path, file) = match create(&self.dir, &stem) { Ok(created) => created, @@ -381,10 +371,48 @@ impl Run { } } +/// The longest file name, in bytes of UTF-8, that common file systems take. +/// Windows counts UTF-16 units instead, which are never more. +const MAX_NAME: usize = 255; + +/// The most [`create`] adds to a stem: the largest number, and the extension. +const SUFFIX: usize = " (4294967295).mp4".len(); + +/// The stem of a recording's program file: the scene file's, cut to fit, and +/// when it started. +fn program_stem(scenes: &str, stamp: &str) -> String { + let budget = (MAX_NAME - SUFFIX).saturating_sub(stamp.len() + 1); + format!("{} {stamp}", truncate(scenes, budget)) +} + +/// The stem of the source `name`'s file beside the program's, `program`, +/// holding as much of its name as fits. +fn source_stem(program: &str, name: &str) -> String { + // The scene file names sources, and a name may hold a path separator. + let safe: String = name + .chars() + .map(|c| match c { + c if c.is_alphanumeric() || " -_.()".contains(c) => c, + _ => '_', + }) + .collect(); + let budget = (MAX_NAME - SUFFIX).saturating_sub(program.len() + 1); + format!("{program} {}", truncate(&safe, budget)) +} + +/// The longest start of `text` within `bytes`, cut between characters. +fn truncate(text: &str, bytes: usize) -> &str { + let mut end = bytes.min(text.len()); + while !text.is_char_boundary(end) { + end -= 1; + } + &text[..end] +} + /// Create a file named `stem` in `dir`, numbering it rather than replacing /// one already there. fn create(dir: &Path, stem: &str) -> anyhow::Result<(PathBuf, std::fs::File)> { - let mut number = 1; + let mut number: u32 = 1; loop { let name = match number { 1 => format!("{stem}.mp4"), @@ -393,7 +421,9 @@ fn create(dir: &Path, stem: &str) -> anyhow::Result<(PathBuf, std::fs::File)> { let path = dir.join(name); match std::fs::File::create_new(&path) { Ok(file) => return Ok((path, file)), - Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => number += 1, + Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => { + number = number.checked_add(1).context("every numbered name is taken")?; + } Err(err) => return Err(err).with_context(|| format!("cannot create {}", path.display())), } } @@ -1013,6 +1043,26 @@ mod tests { assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0); } + /// A long scene file or source name is cut, between characters, so every + /// file name fits, numbered ones included. + #[test] + fn a_long_name_is_cut_to_fit() { + let dir = tempfile::tempdir().unwrap(); + // The first leaves an odd budget for the source's two-byte characters, + // and the second none. + for scenes in ["shows".to_string(), "s".repeat(250)] { + let program = program_stem(&scenes, "2026-11-03T12-00-00Z"); + assert!(program.ends_with(" 2026-11-03T12-00-00Z"), "{program}"); + let source = source_stem(&program, &"é".repeat(300)); + assert!(source.len() > MAX_NAME - SUFFIX - 2, "{} bytes", source.len()); + for stem in [&program, &source, &source] { + let (path, _) = create(dir.path(), stem).unwrap(); + let name = path.file_name().unwrap().len(); + assert!(name <= MAX_NAME, "{name} bytes"); + } + } + } + /// A name already taken, such as by a recording started in the same /// second, is numbered rather than replaced. #[test] From fcb3a99aad0ed3c3f19259cd1afd3da162b0e559 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 22:12:07 -0800 Subject: [PATCH 11/17] fix(studio): a failed catalog leaves each file to report it Following the catalog for sources and rollover stopped the whole recording when the catalog failed, which raced each file's own read of it: a file stopped first was saved or removed as empty, hiding why. Following now just ends, and each file ends with the failure it read. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 22 ++++++++++++++++------ 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index f9b0c9f9..05e8c211 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -243,7 +243,7 @@ impl Run { let mut writers = tokio::task::JoinSet::new(); self.spawn(None, program, file, &mut writers); if let Err(err) = self.follow(&mut writers, sources).await { - tracing::error!(error = format!("{err:#}"), "cannot follow the catalog"); + tracing::error!(error = format!("{err:#}"), "a recording's writer failed"); self.stop.send_replace(true); } while writers.join_next().await.is_some() {} @@ -258,8 +258,13 @@ impl Run { /// file failed while it stays listed gets no other. async fn follow(&self, writers: &mut Writers, sources: bool) -> anyhow::Result<()> { let mut asked = self.stop.subscribe(); + // A catalog that fails ends following it, without stopping the files: + // each reads the same catalog, and its outcome says why. let mut catalog = tokio::select! { - catalog = self.source.catalog::(CatalogFormat::Hang) => catalog?, + catalog = self.source.catalog::(CatalogFormat::Hang) => match catalog { + Ok(catalog) => catalog, + Err(_) => return Ok(()), + }, () = stopped(&mut asked) => return Ok(()), }; let mut listed = BTreeSet::new(); @@ -272,7 +277,7 @@ impl Run { loop { tokio::select! { snapshot = catalog.next() => { - let Some(snapshot) = snapshot? else { return Ok(()) }; + let Ok(Some(snapshot)) = snapshot else { return Ok(()) }; if !sources { continue; } @@ -1027,9 +1032,14 @@ mod tests { recorder.start(false).unwrap(); kio::clock::sleep(Duration::from_millis(100)).await; catalog.abort(moq_net::StreamError::Internal.into()); - kio::clock::timeout(Duration::from_secs(5), recorder.finish()) - .await - .expect("the recording ends with its program"); + // Not stopped, which would only race the failure: it ends on its own. + kio::clock::timeout(Duration::from_secs(5), async { + while !recorder.status().unwrap().done { + kio::clock::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the recording ends with its program"); let status = recorder.status().unwrap(); let [file] = status.files.as_slice() else { From a3b898bf3717fddee5a1abc3f05ab8d62f9595df Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 22:50:13 -0800 Subject: [PATCH 12/17] fix(studio): quitting while recording keeps the program's last pictures The recording stopped before the program's encoders drained on quit, so its file missed what they still held. Quitting now closes the recording: it reads the program until stopping the studio ends the program's tracks, video and now audio too, then finishes, going on in no other file, and the source files finish with it. Co-Authored-By: Claude Opus 5.5 --- README.md | 6 +- rs/moq-studio/src/engine.rs | 58 ++++++++++++++---- rs/moq-studio/src/main.rs | 8 ++- rs/moq-studio/src/record.rs | 116 ++++++++++++++++++++++++++++++++---- 4 files changed, 160 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 98492bb3..fd4ed4d9 100644 --- a/README.md +++ b/README.md @@ -441,8 +441,10 @@ encoders and starts and stops without touching what the relay serves. Without records or watches it. Each file plays while it grows, so a crash leaves a playable fragmented MP4, -and becomes a regular MP4 when the recording stops or the studio quits. A -file nothing reached before the recording stopped is removed. +and becomes a regular MP4 when the recording stops or the studio quits. +Quitting first lets the program's encoders publish what they still hold, and +the program's file keeps it. A file nothing reached before the recording +stopped is removed. ### Source frames diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index dc7ed3c2..7dba92fc 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -430,8 +430,9 @@ pub struct Engine { _origin: Option, /// Set by [`stop`](Self::stop), which ends the program clock. stopped: AtomicBool, - /// The program clock's thread, which [`stop`](Self::stop) waits on. - program_thread: Mutex>>, + /// The program clock's and the audio's threads, which [`stop`](Self::stop) + /// waits on. + threads: Mutex>>, } /// The audio the program publishes. @@ -562,7 +563,7 @@ impl Engine { program, _origin: subscribe, stopped: AtomicBool::new(false), - program_thread: Mutex::default(), + threads: Mutex::default(), }); let Publish { renditions, @@ -581,12 +582,13 @@ impl Engine { let thread = clocked("program".into(), async move || { program.run_program(start, epoch, renditions, rendition).await })?; - *engine.program_thread.lock().unwrap() = Some(thread); + engine.threads.lock().unwrap().push(thread); let mix = engine.clone(); let config = bus.config().clone(); - clocked("audio".into(), async move || { + let thread = clocked("audio".into(), async move || { mix.run_audio(start, epoch, config, mixer, Some(audio)).await })?; + engine.threads.lock().unwrap().push(thread); let save = engine.clone(); clocked("autosave".into(), async move || save.autosave.run(&save.state).await)?; tokio::spawn(engine.clone().grade(health::Sampler::new(health))); @@ -603,14 +605,16 @@ impl Engine { /// Stop the program clock, then wait for each rendition's encoder to /// publish the copies on their way to it and what it still holds, and to /// end its video track, so quitting loses none of the program's last - /// pictures. + /// pictures. The program audio ends its track too, so a recording reading + /// the program reaches the end of each. pub fn stop(&self) { self.stopped.store(true, Ordering::Relaxed); - let thread = self.program_thread.lock().unwrap().take(); - if let Some(thread) = thread - && thread.join().is_err() - { - tracing::error!("the program thread panicked"); + let threads = std::mem::take(&mut *self.threads.lock().unwrap()); + for thread in threads { + let name = thread.thread().name().unwrap_or_default().to_string(); + if thread.join().is_err() { + tracing::error!(name, "the thread panicked"); + } } } @@ -830,7 +834,8 @@ impl Engine { /// Advance the bus as the program clock advances, and publish the program /// mix stamped `epoch` past `start`. /// - /// Runs without a relay too, so the meters move and sources never back up. + /// Runs without a relay too, so the meters move and sources never back up, + /// until the studio [stops](Self::stop), which finishes the audio track. async fn run_audio( &self, start: Instant, @@ -847,6 +852,14 @@ impl Engine { loop { let tick = clock.tick().await; + if self.stopped.load(Ordering::Relaxed) { + if let Some(mut producer) = producer.take() + && let Err(err) = producer.finish() + { + self.stats.fail_audio(err); + } + return; + } // Sample-exact against the program clock, so audio and video stay // aligned however the ticks jitter. let due = tick.saturating_duration_since(start).as_nanos() * rate / 1_000_000_000; @@ -1452,7 +1465,14 @@ mod tests { /// Finish the recording, and return the handler of every track in each /// file, by file. async fn finished(recorder: &record::Recorder) -> Vec> { - kio::clock::timeout(TIMEOUT, recorder.finish()) + recorder.stop(); + done(recorder).await + } + + /// Wait for the recording to finish, and return the handler of every + /// track in each file, by file. + async fn done(recorder: &record::Recorder) -> Vec> { + kio::clock::timeout(TIMEOUT, recorder.done()) .await .expect("the recording finishes"); @@ -1519,6 +1539,18 @@ mod tests { let status = engine.recorder.status().unwrap(); let name = status.files[1].path.file_name().unwrap().to_string_lossy(); assert!(name.ends_with("Z (2).mp4"), "{name}"); + + // Quitting while recording reads the program until stopping ends its + // tracks, into the one file. + engine.recorder.start(false).expect("recording starts"); + let frames = engine.stats.encoded.load(Ordering::Relaxed); + encoded(&engine, frames + 30).await; + engine.recorder.close(); + let stopping = engine.clone(); + tokio::task::spawn_blocking(move || stopping.stop()).await.unwrap(); + let files = done(&engine.recorder).await; + assert_eq!(files, [["soun", "vide"]]); + assert!(engine.stats.error().is_none()); } /// A rendition a resize retired while its encoder was still opening must diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 8b76186e..4d677208 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -186,12 +186,14 @@ fn main() -> anyhow::Result { let Some(engine) = started.get() else { return ran.map(|()| ExitCode::SUCCESS); }; - // A recording still running is finished, so it is a regular MP4. Before - // the program stops, whose ended video would roll it into another file. - runtime.block_on(engine.recorder.finish()); + // A recording still running reads the program until stopping ends its + // tracks, so it holds what the encoders publish on the way out, and then + // finishes as a regular MP4. + engine.recorder.close(); // The edits since the last autosave, so quitting loses none. let saved = engine.save(); engine.stop(); + runtime.block_on(engine.recorder.done()); if let (Err(err), Err(_)) = (&saved, &ran) { tracing::error!(error = format!("{err:#}"), "cannot save on quit"); } diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 05e8c211..a0f17c57 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -41,6 +41,8 @@ pub struct Recorder { struct Session { started: Instant, stop: Arc>, + /// Set by [`Recorder::close`]. + closing: Arc, done: watch::Receiver, files: Files, } @@ -133,6 +135,7 @@ impl Recorder { let program = Arc::new(File::new(path)); let files: Files = Arc::new(Mutex::new(vec![program.clone()])); let stop = Arc::new(watch::channel(false).0); + let closing = Arc::new(AtomicBool::new(false)); let (finished, done) = watch::channel(false); let run = Run { @@ -141,6 +144,7 @@ impl Recorder { stem, files: files.clone(), stop: stop.clone(), + closing: closing.clone(), }; let file = tokio::fs::File::from_std(file); self.runtime.spawn(async move { @@ -151,6 +155,7 @@ impl Recorder { *session = Some(Session { started: kio::clock::now(), stop, + closing, done, files, }); @@ -165,12 +170,32 @@ impl Recorder { } /// Stop recording and wait until every file is finished. + #[cfg(test)] pub async fn finish(&self) { - let mut done = { - let session = self.session.lock().unwrap(); - let Some(session) = session.as_ref() else { return }; - session.stop.send_replace(true); - session.done.clone() + self.stop(); + self.done().await; + } + + /// Let the recording read the program until the program's own tracks end, + /// as stopping the studio ends them, and then finish, rather than stop where + /// it is or go on in another file. So a recording still running when the + /// studio quits holds what the encoders publish on the way out. + pub fn close(&self) { + if let Some(session) = self.session.lock().unwrap().as_ref() { + session.closing.store(true, Ordering::Relaxed); + } + } + + /// Wait until the recording, if any, has finished every file. + pub async fn done(&self) { + let Some(mut done) = self + .session + .lock() + .unwrap() + .as_ref() + .map(|session| session.done.clone()) + else { + return; }; // Fails only once the recording's task is gone, having finished. let _ = done.wait_for(|done| *done).await; @@ -223,6 +248,8 @@ struct Run { stem: String, files: Files, stop: Arc>, + /// The recording reads the program to its end; see [`Recorder::close`]. + closing: Arc, } /// A recording's file writers, each ending with the file it wrote. @@ -246,7 +273,12 @@ impl Run { tracing::error!(error = format!("{err:#}"), "a recording's writer failed"); self.stop.send_replace(true); } - while writers.join_next().await.is_some() {} + // Whatever ends the program's file ends the recording. + while let Some(joined) = writers.join_next().await { + if matches!(joined, Ok(Ended { source: None, .. })) { + self.stop.send_replace(true); + } + } } /// Go on in a new file whenever the program's tracks are replaced, and with @@ -305,7 +337,7 @@ impl Run { } // A new canvas size or frame rate moves the program video to a // new track, which goes on in the next file. - Ended { source: None, replaced: true } if !*asked.borrow() => { + Ended { source: None, replaced: true } if !*asked.borrow() && !self.closing.load(Ordering::Relaxed) => { if !self.file(None, writers) { self.stop.send_replace(true); return Ok(()); @@ -319,8 +351,8 @@ impl Run { }, () = stopped(&mut asked) => return Ok(()), } - if *asked.borrow() { - return Ok(()); + if *asked.borrow() || self.closing.load(Ordering::Relaxed) { + continue; } let start: Vec = listed .iter() @@ -368,8 +400,9 @@ impl Run { fn spawn(&self, source: Option, entry: Arc, file: tokio::fs::File, writers: &mut Writers) { let broadcast = self.source.clone(); let asked = self.stop.subscribe(); + let closing = self.closing.clone(); writers.spawn(async move { - let (finished, replaced) = write(broadcast, source.as_deref(), file, &entry, asked).await; + let (finished, replaced) = write(broadcast, source.as_deref(), file, &entry, asked, closing).await; entry.finish(finished); Ended { source, replaced } }); @@ -444,6 +477,7 @@ async fn write( file: tokio::fs::File, entry: &File, mut stop: watch::Receiver, + closing: Arc, ) -> (anyhow::Result, bool) { let catalog = tokio::select! { catalog = source.catalog::(CatalogFormat::Hang) => catalog.context("the studio's catalog"), @@ -459,6 +493,7 @@ async fn write( name: name.map(str::to_string), last: None, replaced: replaced.clone(), + closing, }; let finished = record(mp4::Export::new(source, listing), file, entry, stop).await; (finished, replaced.load(Ordering::Relaxed)) @@ -547,6 +582,9 @@ struct Listing { last: Option, /// Set as the listing ends because its tracks were replaced. replaced: Arc, + /// Set by [`Recorder::close`]: the program's listing then ends once the + /// tracks it recorded have all ended, adding and replacing none. + closing: Arc, } impl Stream for Listing { @@ -570,6 +608,26 @@ impl Stream for Listing { } continue; }; + if let Some(last) = &self.last + && self.name.is_none() + && self.closing.load(Ordering::Relaxed) + { + // Each track a publisher ended leaves the catalog, and the export + // reads every one to its end before it finishes. + let listed = |track: &String| { + listing.video.renditions.contains_key(track) || listing.audio.renditions.contains_key(track) + }; + if !last + .video + .renditions + .keys() + .chain(last.audio.renditions.keys()) + .any(listed) + { + return Poll::Ready(Ok(None)); + } + continue; + } if let Some(last) = &self.last { let video = replaced(&last.video.renditions, &listing.video.renditions); let audio = replaced(&last.audio.renditions, &listing.audio.renditions); @@ -944,6 +1002,44 @@ mod tests { } } + /// Closed as the studio quits, a recording reads the program until the + /// program's track ends, rather than stopping where it was, and goes on in + /// no other file even as a new track appears; the sources finish with it. + #[tokio::test(start_paused = true)] + async fn a_closed_recording_reads_the_program_to_its_end() { + let dir = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + studio.add("mic"); + let recorder = studio.recorder(dir.path()); + + recorder.start(true).unwrap(); + studio.play(Duration::from_secs(1)).await; + recorder.close(); + studio.play(Duration::from_secs(1)).await; + studio.replace_program(); + kio::clock::timeout(Duration::from_secs(5), async { + tokio::join!(recorder.done(), studio.play(Duration::from_millis(500))) + }) + .await + .expect("the recording finishes once the program's track ends"); + + let status = recorder.status().unwrap(); + let files: Vec<_> = status + .files + .iter() + .map(|file| { + assert_eq!(file.finished, Some(Finished::Saved), "{}", file.path.display()); + let name = file.path.file_name().unwrap().to_string_lossy().into_owned(); + let [(_, duration)] = tracks(&file.path).try_into().unwrap(); + (name.rsplit_once("Z").unwrap().1.to_string(), duration) + }) + .collect(); + let names: Vec<_> = files.iter().map(|(name, _)| name.as_str()).collect(); + assert_eq!(names, [".mp4", " mic.mp4"]); + let program = files[0].1; + assert!(about(program, Duration::from_secs(2)), "{files:?}"); + } + /// A source whose file cannot be created fails without stopping the rest, /// and once removed and added back, records again. #[cfg(unix)] From c79470bd5c189fb3679229567440f33ab30a2b14 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 22:53:24 -0800 Subject: [PATCH 13/17] fix(studio): a recording closed before reading anything still finishes Closing waited for the tracks the program file recorded to end, but one whose first snapshot came after the studio stopped had recorded none, so it waited for a catalog update that never came and quitting hung. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 57 +++++++++++++++++++++++++++++++------ 1 file changed, 48 insertions(+), 9 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index a0f17c57..48a8baf6 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -608,12 +608,17 @@ impl Stream for Listing { } continue; }; - if let Some(last) = &self.last - && self.name.is_none() - && self.closing.load(Ordering::Relaxed) - { + if self.name.is_none() && self.closing.load(Ordering::Relaxed) { // Each track a publisher ended leaves the catalog, and the export - // reads every one to its end before it finishes. + // reads every one to its end before it finishes. Before the first + // snapshot, the tracks listed are the ones left to read. + let Some(last) = &self.last else { + if listing.video.renditions.is_empty() && listing.audio.renditions.is_empty() { + return Poll::Ready(Ok(None)); + } + self.last = Some(listing.clone()); + return Poll::Ready(Ok(Some(listing))); + }; let listed = |track: &String| { listing.video.renditions.contains_key(track) || listing.audio.renditions.contains_key(track) }; @@ -709,7 +714,8 @@ mod tests { origin: moq_net::origin::Producer, publication: moq_net::origin::Publication, catalog: moq_scene::publish::Catalog, - program: Tone, + /// `None` once [ended](Self::end_program). + program: Option, /// Each listed source's tone, and the scratch broadcast its own catalog /// writes into. sources: Vec<(String, Tone, moq_net::broadcast::Producer, moq_mux::catalog::Producer)>, @@ -724,7 +730,7 @@ mod tests { let mut broadcast = publication.broadcast().clone(); let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); let catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); - let program = Tone::new(&mut broadcast, catalog.clone()); + let program = Some(Tone::new(&mut broadcast, catalog.clone())); publication.announce(Default::default()).unwrap(); Self { origin, @@ -783,7 +789,7 @@ mod tests { let mut broadcast = self.publication.broadcast().clone(); let old = self.catalog.snapshot().audio.renditions; let tone = Tone::new(&mut broadcast, self.catalog.clone()); - let mut old_tone = std::mem::replace(&mut self.program, tone); + let mut old_tone = self.program.replace(tone).unwrap(); old_tone.audio.finish().unwrap(); drop(old_tone); self.catalog @@ -791,6 +797,17 @@ mod tests { .unwrap(); } + /// End the program's track, as stopping the studio does. + fn end_program(&mut self) { + let old = self.catalog.snapshot().audio.renditions; + let mut tone = self.program.take().unwrap(); + tone.audio.finish().unwrap(); + drop(tone); + self.catalog + .mutate(|root| root.audio.renditions.retain(|track, _| !old.contains_key(track))) + .unwrap(); + } + /// List what each source's own catalog holds, as the studio mirrors it. fn relist(&mut self) { let listings: Vec<_> = self @@ -811,7 +828,9 @@ mod tests { async fn play(&mut self, duration: Duration) { let end = self.played + duration; while self.played < end { - self.program.write(self.played); + if let Some(program) = &mut self.program { + program.write(self.played); + } for (_, tone, ..) in &mut self.sources { tone.write(self.played); } @@ -1040,6 +1059,26 @@ mod tests { assert!(about(program, Duration::from_secs(2)), "{files:?}"); } + /// Closed and its program ended before its file read anything, a recording + /// still finishes, with nothing left to wait for. + #[tokio::test(start_paused = true)] + async fn a_recording_closed_before_it_read_anything_finishes() { + let dir = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + let recorder = studio.recorder(dir.path()); + recorder.start(false).unwrap(); + recorder.close(); + studio.end_program(); + kio::clock::timeout(Duration::from_secs(5), recorder.done()) + .await + .expect("the recording finishes without another catalog update"); + let status = recorder.status().unwrap(); + let [file] = status.files.as_slice() else { + panic!("{} files", status.files.len()); + }; + assert_eq!(file.finished, Some(Finished::Empty)); + } + /// A source whose file cannot be created fails without stopping the rest, /// and once removed and added back, records again. #[cfg(unix)] From 4dbcfdb8731efaa6cd05cc63842d4b2fb2ebebe8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 23:39:49 -0800 Subject: [PATCH 14/17] fix(studio): closing wakes a recording, and a waiting failure beats a stop A program file that had read a catalog with no program tracks waited on the catalog when closed, which no longer changes, so quitting hung; the listing now watches the close too and ends with nothing left to read. A stop that raced a failure already waiting could report the file empty or saved; the file now takes what the export has, the failure included, before the stop. The on-demand source test also covers the next subscriber after the gap. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/publish.rs | 10 ++++ rs/moq-studio/src/record.rs | 98 +++++++++++++++++++++++++++++++++---- 2 files changed, 98 insertions(+), 10 deletions(-) diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 29bf11af..55935f4f 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -740,6 +740,7 @@ mod tests { assert!(video.track.is_open(), "a subscriber gets pictures"); let mut group = track.recv_group().await.unwrap().expect("a group"); assert!(group.read_frame().await.unwrap().is_some()); + let first = group.sequence; drop(group); drop(track); @@ -748,6 +749,15 @@ mod tests { } video.encode(&mut gray(size, 66)).unwrap(); assert!(!video.track.is_open(), "the encoder closes once nobody subscribes"); + + // The next subscriber gets a fresh encoder's pictures, in a group + // after the gap rather than the one before it. + let mut track = consumer.track(&name).unwrap().subscribe(None).await.unwrap(); + video.encode(&mut gray(size, 100)).unwrap(); + assert!(video.track.is_open(), "a new subscriber reopens the encoder"); + let mut group = track.recv_group().await.unwrap().expect("a group"); + assert!(group.sequence > first, "{} after {first}", group.sequence); + assert!(group.read_frame().await.unwrap().is_some()); } => {} } }) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 48a8baf6..2cc781ab 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -42,7 +42,7 @@ struct Session { started: Instant, stop: Arc>, /// Set by [`Recorder::close`]. - closing: Arc, + closing: kio::Producer, done: watch::Receiver, files: Files, } @@ -135,7 +135,7 @@ impl Recorder { let program = Arc::new(File::new(path)); let files: Files = Arc::new(Mutex::new(vec![program.clone()])); let stop = Arc::new(watch::channel(false).0); - let closing = Arc::new(AtomicBool::new(false)); + let closing = kio::Producer::new(false); let (finished, done) = watch::channel(false); let run = Run { @@ -144,7 +144,7 @@ impl Recorder { stem, files: files.clone(), stop: stop.clone(), - closing: closing.clone(), + closing: closing.consume(), }; let file = tokio::fs::File::from_std(file); self.runtime.spawn(async move { @@ -182,7 +182,10 @@ impl Recorder { /// studio quits holds what the encoders publish on the way out. pub fn close(&self) { if let Some(session) = self.session.lock().unwrap().as_ref() { - session.closing.store(true, Ordering::Relaxed); + // Fails only once the recording has finished. + if let Ok(mut closing) = session.closing.write() { + *closing = true; + } } } @@ -249,7 +252,7 @@ struct Run { files: Files, stop: Arc>, /// The recording reads the program to its end; see [`Recorder::close`]. - closing: Arc, + closing: kio::Consumer, } /// A recording's file writers, each ending with the file it wrote. @@ -337,7 +340,7 @@ impl Run { } // A new canvas size or frame rate moves the program video to a // new track, which goes on in the next file. - Ended { source: None, replaced: true } if !*asked.borrow() && !self.closing.load(Ordering::Relaxed) => { + Ended { source: None, replaced: true } if !*asked.borrow() && !*self.closing.read() => { if !self.file(None, writers) { self.stop.send_replace(true); return Ok(()); @@ -351,7 +354,7 @@ impl Run { }, () = stopped(&mut asked) => return Ok(()), } - if *asked.borrow() || self.closing.load(Ordering::Relaxed) { + if *asked.borrow() || *self.closing.read() { continue; } let start: Vec = listed @@ -477,9 +480,11 @@ async fn write( file: tokio::fs::File, entry: &File, mut stop: watch::Receiver, - closing: Arc, + closing: kio::Consumer, ) -> (anyhow::Result, bool) { let catalog = tokio::select! { + // A failure already there is reported rather than lost to the stop. + biased; catalog = source.catalog::(CatalogFormat::Hang) => catalog.context("the studio's catalog"), () = stopped(&mut stop) => return (remove(file, entry).await, false), }; @@ -513,6 +518,8 @@ async fn record( let mut stopping = false; let ended = loop { let next = tokio::select! { + // What the export already has, a failure included, before the stop. + biased; next = export.next() => next, // What was read is still written. () = stopped(&mut stop), if !stopping => { @@ -584,7 +591,7 @@ struct Listing { replaced: Arc, /// Set by [`Recorder::close`]: the program's listing then ends once the /// tracks it recorded have all ended, adding and replacing none. - closing: Arc, + closing: kio::Consumer, } impl Stream for Listing { @@ -594,6 +601,25 @@ impl Stream for Listing { if self.replaced.load(Ordering::Relaxed) { return Poll::Ready(Ok(None)); } + // Polled for the program's listing, so closing wakes it: with nothing + // left to read, it ends without waiting for another catalog update. + let closing = self.name.is_none() + && matches!( + self.closing.poll(waiter, |closing| if **closing { + Poll::Ready(()) + } else { + Poll::Pending + }), + Poll::Ready(Ok(())) + ); + if closing + && self + .last + .as_ref() + .is_some_and(|last| last.video.renditions.is_empty() && last.audio.renditions.is_empty()) + { + return Poll::Ready(Ok(None)); + } loop { let Some(studio) = ready!(self.catalog.poll_next(waiter))? else { return Poll::Ready(Ok(None)); @@ -608,7 +634,7 @@ impl Stream for Listing { } continue; }; - if self.name.is_none() && self.closing.load(Ordering::Relaxed) { + if closing { // Each track a publisher ended leaves the catalog, and the export // reads every one to its end before it finishes. Before the first // snapshot, the tracks listed are the ones left to read. @@ -1079,6 +1105,58 @@ mod tests { assert_eq!(file.finished, Some(Finished::Empty)); } + /// A recording that read the program with no tracks left, such as one whose + /// video never opened and whose audio failed, finishes once closed, without + /// waiting for a catalog update that will not come. + #[tokio::test(start_paused = true)] + async fn a_recording_of_an_empty_program_finishes_once_closed() { + let dir = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + studio.end_program(); + let recorder = studio.recorder(dir.path()); + recorder.start(false).unwrap(); + kio::clock::sleep(Duration::from_millis(500)).await; + assert!(!recorder.status().unwrap().done, "nothing ends it before closing"); + recorder.close(); + kio::clock::timeout(Duration::from_secs(5), recorder.done()) + .await + .expect("closing wakes the recording"); + let status = recorder.status().unwrap(); + let [file] = status.files.as_slice() else { + panic!("{} files", status.files.len()); + }; + assert_eq!(file.finished, Some(Finished::Empty)); + } + + /// A failure already waiting as a recording stops is what each file + /// reports, the program's and every source's, rather than lost to the stop. + #[tokio::test(start_paused = true)] + async fn a_failure_wins_over_a_stop() { + let dir = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + studio.add("mic"); + let recorder = studio.recorder(dir.path()); + recorder.start(true).unwrap(); + studio.play(Duration::from_secs(1)).await; + // Together, with no chance for either file to read the failure first. + studio.catalog.clone().abort(moq_net::StreamError::Internal.into()); + recorder.stop(); + kio::clock::timeout(Duration::from_secs(5), recorder.done()) + .await + .expect("the recording finishes"); + + let status = recorder.status().unwrap(); + assert_eq!(status.files.len(), 2); + for file in &status.files { + assert!( + matches!(&file.finished, Some(Finished::Failed(reason)) if reason.contains("reading the broadcast")), + "{}: {:?}", + file.path.display(), + file.finished + ); + } + } + /// A source whose file cannot be created fails without stopping the rest, /// and once removed and added back, records again. #[cfg(unix)] From 77bfd6ff186028caee8a630478262535c98db3de Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 23:45:14 -0800 Subject: [PATCH 15/17] fix(studio): a stop is never held off by media that keeps arriving Letting a ready export win over the stop meant a disk slower than the broadcast could keep a recording going after Stop. The stop is checked on every chunk instead, with one look at the export first so a failure already waiting is still what the file reports. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 24 +++++++++++++++--------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 2cc781ab..a9a3cff5 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -517,15 +517,21 @@ async fn record( let mut export = export.with_max_delay(MAX_DELAY); let mut stopping = false; let ended = loop { - let next = tokio::select! { - // What the export already has, a failure included, before the stop. - biased; - next = export.next() => next, - // What was read is still written. - () = stopped(&mut stop), if !stopping => { - export.stop(); - stopping = true; - continue; + let next = if !stopping && *stop.borrow() { + // Checked on every chunk, so media that keeps arriving never holds a + // stop off. One look first, which reports a failure already waiting + // rather than losing it to the stop; what was read is still written. + stopping = true; + let next = export.poll_next(&kio::Waiter::noop()); + export.stop(); + match next { + Poll::Ready(next) => next, + Poll::Pending => continue, + } + } else { + tokio::select! { + next = export.next() => next, + () = stopped(&mut stop), if !stopping => continue, } }; match next { From 448e11c59365aaf291c6b555d7c033b887f36116 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 00:52:40 -0800 Subject: [PATCH 16/17] fix(studio): a closed empty recording reports a catalog that just failed Closing ended a program file with nothing left to read before polling the catalog, so a catalog that had just failed went unread and the file reported empty. The catalog is polled first, and the file ends empty only once nothing is waiting there. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 66 +++++++++++++++++++++++-------------- 1 file changed, 42 insertions(+), 24 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index a9a3cff5..85f59dba 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -11,7 +11,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; -use std::task::{Poll, ready}; +use std::task::Poll; use std::time::Duration; use anyhow::Context; @@ -607,8 +607,7 @@ impl Stream for Listing { if self.replaced.load(Ordering::Relaxed) { return Poll::Ready(Ok(None)); } - // Polled for the program's listing, so closing wakes it: with nothing - // left to read, it ends without waiting for another catalog update. + // Polled for the program's listing, so closing wakes it. let closing = self.name.is_none() && matches!( self.closing.poll(waiter, |closing| if **closing { @@ -618,16 +617,22 @@ impl Stream for Listing { }), Poll::Ready(Ok(())) ); - if closing - && self - .last - .as_ref() - .is_some_and(|last| last.video.renditions.is_empty() && last.audio.renditions.is_empty()) - { - return Poll::Ready(Ok(None)); - } loop { - let Some(studio) = ready!(self.catalog.poll_next(waiter))? else { + let studio = match self.catalog.poll_next(waiter) { + Poll::Ready(studio) => studio?, + // With nothing new, a closing program listing with nothing left to + // read ends without waiting for another catalog update. A catalog + // that failed is reported above instead. + Poll::Pending => { + let empty = + |last: &hang::Catalog| last.video.renditions.is_empty() && last.audio.renditions.is_empty(); + if closing && self.last.as_ref().is_some_and(empty) { + return Poll::Ready(Ok(None)); + } + return Poll::Pending; + } + }; + let Some(studio) = studio else { return Poll::Ready(Ok(None)); }; let listing = match &self.name { @@ -1113,25 +1118,38 @@ mod tests { /// A recording that read the program with no tracks left, such as one whose /// video never opened and whose audio failed, finishes once closed, without - /// waiting for a catalog update that will not come. + /// waiting for a catalog update that will not come. A catalog that failed + /// just before is what it reports. #[tokio::test(start_paused = true)] async fn a_recording_of_an_empty_program_finishes_once_closed() { let dir = tempfile::tempdir().unwrap(); let mut studio = Studio::new(); studio.end_program(); let recorder = studio.recorder(dir.path()); - recorder.start(false).unwrap(); - kio::clock::sleep(Duration::from_millis(500)).await; - assert!(!recorder.status().unwrap().done, "nothing ends it before closing"); - recorder.close(); - kio::clock::timeout(Duration::from_secs(5), recorder.done()) - .await - .expect("closing wakes the recording"); - let status = recorder.status().unwrap(); - let [file] = status.files.as_slice() else { - panic!("{} files", status.files.len()); + let file = async |abort: bool| { + recorder.start(false).unwrap(); + kio::clock::sleep(Duration::from_millis(500)).await; + assert!(!recorder.status().unwrap().done, "nothing ends it before closing"); + if abort { + // With no chance for the file to read the failure before closing. + studio.catalog.clone().abort(moq_net::StreamError::Internal.into()); + } + recorder.close(); + kio::clock::timeout(Duration::from_secs(5), recorder.done()) + .await + .expect("closing wakes the recording"); + let status = recorder.status().unwrap(); + let [file] = status.files.as_slice() else { + panic!("{} files", status.files.len()); + }; + file.finished.clone() }; - assert_eq!(file.finished, Some(Finished::Empty)); + assert_eq!(file(false).await, Some(Finished::Empty)); + let failed = file(true).await; + assert!( + matches!(&failed, Some(Finished::Failed(reason)) if reason.contains("reading the broadcast")), + "{failed:?}" + ); } /// A failure already waiting as a recording stops is what each file From 1967201d53a105ab237fd744877acf54ceb72609 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 01:58:19 -0800 Subject: [PATCH 17/17] fix(studio): a long scene file name leaves room for each source's name A scene stem of 216 bytes or more left source files no room for their source's name, so they differed only by number. The scene stem is now cut to leave each source at least 64 bytes. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 85f59dba..035b1c75 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -419,10 +419,14 @@ const MAX_NAME: usize = 255; /// The most [`create`] adds to a stem: the largest number, and the extension. const SUFFIX: usize = " (4294967295).mp4".len(); -/// The stem of a recording's program file: the scene file's, cut to fit, and -/// when it started. +/// What a long scene file name leaves of each source's file name, in bytes, +/// so the Output panel can still tell the source files apart. +const SOURCE: usize = 64; + +/// The stem of a recording's program file: the scene file's, cut to fit +/// beside a source's name, and when it started. fn program_stem(scenes: &str, stamp: &str) -> String { - let budget = (MAX_NAME - SUFFIX).saturating_sub(stamp.len() + 1); + let budget = (MAX_NAME - SUFFIX).saturating_sub(stamp.len() + 1 + SOURCE + 1); format!("{} {stamp}", truncate(scenes, budget)) } @@ -1296,12 +1300,13 @@ mod tests { fn a_long_name_is_cut_to_fit() { let dir = tempfile::tempdir().unwrap(); // The first leaves an odd budget for the source's two-byte characters, - // and the second none. + // and the second only the source's reserve. for scenes in ["shows".to_string(), "s".repeat(250)] { let program = program_stem(&scenes, "2026-11-03T12-00-00Z"); assert!(program.ends_with(" 2026-11-03T12-00-00Z"), "{program}"); let source = source_stem(&program, &"é".repeat(300)); assert!(source.len() > MAX_NAME - SUFFIX - 2, "{} bytes", source.len()); + assert!(source.len() >= program.len() + 1 + SOURCE - 1, "{source}"); for stem in [&program, &source, &source] { let (path, _) = create(dir.path(), stem).unwrap(); let name = path.file_name().unwrap().len();