diff --git a/Cargo.lock b/Cargo.lock index 0ac89ad..f61e069 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3829,6 +3829,7 @@ dependencies = [ "egui", "egui_dock", "hang", + "humantime", "kio", "moq-audio", "moq-auth", diff --git a/Cargo.toml b/Cargo.toml index b4cb309..1d70980 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,6 +41,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 a810125..e05ffbd 100644 --- a/README.md +++ b/README.md @@ -16,8 +16,10 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ - `show.toml` is the scene collection. If it does not exist yet, [guided setup](#first-launch) proposes one and creates it. 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. - `--failover ` publishes the same broadcast to a second relay under the same broadcast instance, so a viewer linked to both resumes mid-group from it when the first fails. Both relays must speak the one `--connect-version`, @@ -97,7 +99,7 @@ Every panel is a tab you can drag, split, or stack. | Guests | Invite a guest, show their join link as a QR code, see who joined, and take a guest out of every scene. | | Filters | Crop the selected source or item, add, order, and set its filters, set an item's blend mode, and copy or paste filters. | | Audio | Set each audio source's gain and mute, and watch its meter and the program's. | -| Output | Set the canvas size and frame rate, and see the broadcast path, whether the program is encoding, and the encoder's work. | +| Output | Start and stop recording, set the canvas size and frame rate, and see the broadcast path, whether the program is encoding, and the encoder's work. | | Health | See how the stream reaches viewers, relay by relay and track by track, and the cause of each problem. | | Replay | Replay the last moments of the program or a source in a replay source, slowed down if you like, or save them as a clip. | @@ -382,12 +384,13 @@ rendition's track ends and a new one is listed in its place in a catalog update, on the same timeline, and viewers switch to it; the video pauses while the new encoder opens. Items keep their place in canvas pixels. A canvas the encoder cannot take stops the video, saying why, until the canvas changes -again. +again. At startup, it stops the studio when live; without `--connect`, the +studio starts without video. The program encodes whether or not anyone watches, so a viewer who arrives late can fetch the recent past. `--on-demand` copies and encodes the video only -while someone watches it, which saves that work but leaves nothing from before -the first viewer; audio encodes either way. A hardware encoder that reads +while someone watches or records it, which saves that work but leaves nothing +from before the first viewer; audio encodes either way. A hardware encoder that reads the GPU (NVENC on Linux, VideoToolbox on macOS, Media Foundation on Windows) takes each frame there when it can read the GPU the studio draws on. When it cannot, such as NVENC while the studio draws on a laptop's integrated GPU, the @@ -420,7 +423,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 @@ -439,6 +443,33 @@ 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. 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 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 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 +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. +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 Sources make the opposite trip. A frame already on the GPU is drawn where it diff --git a/justfile b/justfile index 3697418..61f640f 100644 --- a/justfile +++ b/justfile @@ -55,7 +55,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) | test(=setup::tests::times_every_canvas)' + cargo nextest run --locked -p moq-studio --features openh264 --run-ignored only -E 'test(/^engine::tests::the_program_/) | test(=setup::tests::times_every_canvas)' # CEF's browser process opens an X display even to paint offscreen. xvfb=$(nix build --inputs-from . --no-link --print-out-paths nixpkgs#xvfb-run) "$xvfb/bin/xvfb-run" -a cargo nextest run --locked -p moq-studio --features cef --run-ignored only -E 'binary(browser)' diff --git a/quest/m0/README.md b/quest/m0/README.md index 5182a66..8a2c12a 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -104,7 +104,6 @@ repository. - [Guest return feed](/quest/m0/guest-return.md) - guests see the program and hear a mix-minus - [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 on macOS and Windows](/quest/m0/browser-source.md) - the out-of-process CEF pages Linux has, with shared GPU textures on the other desktops - [Browser sources keep a profile](/quest/m0/browser-profile.md) - one persistent profile, so logins survive restarts, and a reload action - [Browser source popups](/quest/m0/browser-popups.md) - dropdowns and other in-page popups draw and take input diff --git a/quest/m0/recording.md b/quest/m0/recording.md deleted file mode 100644 index 5ccb113..0000000 --- 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/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 90e4aa5..55935f4 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>, } @@ -88,10 +90,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 + } + /// What each source track the studio catalog lists has written, keyed by /// rendition ID, as [`Catalog::stats`](moq_mux::catalog::Producer::stats) /// reports the program's. The broadcast names every track uniquely, so all @@ -151,6 +162,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, @@ -172,6 +184,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, @@ -277,6 +290,12 @@ impl Encoding { self.encoder.is_some() } + /// Whether anyone subscribes to the track. + pub(crate) fn wanted(&self) -> bool { + let producer = self.producer.as_ref().expect("open until it ends"); + producer.demand().wanted().is_some() + } + /// Encode with `encoder` from here on, after draining the one it replaces. pub fn open(&mut self, encoder: E) -> Result<(), Error> { self.drain()?; @@ -365,6 +384,8 @@ impl Drop for Encoding { pub(crate) struct Video { track: Encoding, config: moq_video::encode::Config, + /// 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. @@ -395,14 +416,16 @@ impl Video { Ok(Self { track: Encoding::new(producer), config, + on_demand: listing.on_demand, clock: listing.studio.clock(), epoch: None, }) } - /// Encode and publish `frame`, restamping it onto the studio clock. The - /// encoder opens on the first frame, and reopens when the picture's size or - /// color space changes, which moves the rendition to a new track. + /// Encode and publish `frame`, restamping it onto the studio clock, unless + /// it is on demand and nobody subscribes. The encoder opens on the first + /// frame it encodes, and reopens when the picture's size or color space + /// changes, which moves the rendition to a new track. pub(crate) fn encode(&mut self, frame: &mut moq_video::Frame) -> Result<(), Error> { let captured = frame.timestamp.as_micros(); let epoch = *self @@ -410,6 +433,15 @@ 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)?; + if self.on_demand && !self.track.wanted() { + // 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.track.is_open() { + self.track.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) { @@ -678,6 +710,60 @@ 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.track.is_open(), "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.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); + for _ in 0..100 { + tokio::task::yield_now().await; + } + 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()); + } => {} + } + }) + .await; + } + /// Holds its newest pictures back the way a pipelined hardware encoder /// does, handing them over only on a flush or finish. openh264, the encoder /// on every machine, holds nothing. diff --git a/rs/moq-studio/Cargo.toml b/rs/moq-studio/Cargo.toml index b7f4918..02bc7da 100644 --- a/rs/moq-studio/Cargo.toml +++ b/rs/moq-studio/Cargo.toml @@ -47,6 +47,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"] } # Signs guest tokens with the room key. diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index fe9f37f..7dba92f 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -1,6 +1,7 @@ //! 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 relays serve +//! that broadcast, and recordings read it. //! //! Threads: //! - **program** draws preview and program once per canvas frame and hands the @@ -9,6 +10,7 @@ //! program mix. //! - **encode**: one per program rendition, which waits for each program copy //! and encodes it. A new canvas size or frame rate starts a new rendition. +//! - 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 @@ -16,7 +18,7 @@ //! - The UI runs on the main thread and edits [`State`] under its lock. use std::collections::BTreeMap; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; @@ -30,7 +32,7 @@ use moq_scene::{Canvas, Collection, Compositor, Draft, Edit, 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. @@ -265,7 +267,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, /// Whether the program is handed to an encoder now: not while on demand /// and unwatched, nor while a new rendition opens. @@ -336,21 +339,16 @@ impl Stats { } } -/// Where the program goes. +/// The relays the program goes live on. pub struct Output { - /// The broadcast path on the relay. + /// The broadcast path on the relays. 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, /// A connection to each relay, which reconnects on its own and dials only /// while held. The Health panel reads their states. _connections: Vec, - publication: moq_net::origin::Publication, - /// What the broadcast lists and the clock its timestamps are on, for a - /// replay of the program or a source. - catalog: moq_scene::publish::Catalog, - clock: moq_mux::Clock, } /// A relay the program is published to. @@ -360,13 +358,25 @@ pub struct Relay { pub url: url::Url, } -impl Drop for Output { +/// The program's broadcast, open for as long as the engine runs. +pub struct Program { + /// Where the broadcast is announced: the relay connections publish it, + /// and recordings read it. + origin: moq_net::origin::Producer, + publication: moq_net::origin::Publication, + /// What the broadcast lists and the clock its timestamps are on, for a + /// replay of the program or a source. + catalog: moq_scene::publish::Catalog, + clock: moq_mux::Clock, +} + +impl Drop for Program { fn drop(&mut self) { self.publication.close(); } } -impl Output { +impl Program { /// What a replay of the program copies: its video and audio, up to now. pub fn program(&self) -> moq_scene::replay::Target { let catalog = self.catalog.snapshot(); @@ -410,14 +420,19 @@ pub struct Engine { pub mix: moq_audio::bus::Mix, /// `None` when running without a relay. pub output: Option, + /// Records the program, and each source too if asked. + pub recorder: record::Recorder, /// The room guests join, when configured. pub room: Option>, + /// The program's broadcast. + pub program: Program, /// Keeps broadcast sources' origin open when no connection holds it. _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: Mutex>>, + /// The program clock's and the audio's threads, which [`stop`](Self::stop) + /// waits on. + threads: Mutex>>, } /// The audio the program publishes. @@ -436,8 +451,10 @@ pub struct Launch { pub relays: Vec, /// The broadcast path the program is published at. pub broadcast: String, - /// Whether the program video encodes only while someone watches. + /// Whether the program video encodes only while someone watches or records. pub on_demand: bool, + /// The directory recordings are saved in. + pub recordings: PathBuf, /// The room guests join, if any. pub room: Option>, } @@ -446,10 +463,11 @@ impl Engine { /// Open the `loaded` collection and history on `launch.gpu`, start the /// program clock, and save both to `autosave`'s files as they change, /// starting at once. - /// Publish the program at `launch.broadcast` on every one of - /// `launch.relays`, preferred first, encoding its video only while someone - /// watches when `launch.on_demand`, and add the guests who join `launch.room` - /// as sources. + /// The program is encoded into a broadcast at `launch.broadcast` while + /// anyone wants it: a recording into `launch.recordings`, or every one of + /// `launch.relays`, preferred first, which want it all the time unless + /// `launch.on_demand`. The guests who join `launch.room` are added as + /// sources. /// /// Call inside the tokio runtime that serves the relay connections. pub async fn start(launch: Launch, autosave: Autosave, loaded: autosave::Loaded) -> anyhow::Result> { @@ -458,10 +476,11 @@ impl Engine { relays, broadcast, on_demand, + recordings, room, } = launch; let (bus, mixer) = moq_audio::bus::Bus::new(Default::default())?; - let program = bus.mix(); + let program_mix = bus.mix(); let subscribe = moq_tokio::origin::spawn(); // Every relay feeds `subscribe`, so a guest either one routes is seen. let guests = match &room { @@ -482,23 +501,41 @@ impl Engine { let canvas = collection.canvas; let stats = Arc::new(Stats::default()); - let (output, origin, publish) = if relays.is_empty() { - (None, Some(subscribe), None) + let path = + moq_net::Path::new(broadcast.as_str()).with_context(|| format!("cannot publish at `{broadcast}`"))?; + let origin = moq_tokio::origin::spawn(); + let (live, subscribe) = if relays.is_empty() { + (None, Some(subscribe)) } else { - let (output, publish) = publish(&gpu, &stats, relays, subscribe, &broadcast, canvas, on_demand).await?; - (Some(output), None, Some(publish)) + tracing::info!(broadcast = %path, "publishing"); + let reports = subscribe.consume(); + let (connections, links) = connect(relays, &origin, &subscribe); + let live = Live { + connections, + links, + reports, + }; + (Some(live), None) }; + let (program, publish) = publish(&gpu, &stats, origin, &path, canvas, live.as_ref()).await?; + let source = moq_mux::Source::new(program.origin.consume(), program.publication.broadcast().consume()); + let recorder = record::Recorder::new(source, &recordings, autosave.path())?; + let output = live.map(|live| Output { + path: path.to_string(), + on_demand, + _connections: live.connections, + }); // 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()); let dir = std::path::absolute(dir.unwrap_or(Path::new(".")))?; let sources = Sources::new(source::Context { bus: bus.clone(), - program: program.clone(), + program: program_mix.clone(), origin: sources, dir, output: moq_video::render::output(&gpu.device), - publish: publish.as_ref().map(|publish| publish.sources.clone()), + publish: Some(publish.sources.clone()), browser: browser(&gpu)?, }); let mut compositor = Compositor::new(&gpu.device, &gpu.queue, collection.canvas)?; @@ -519,45 +556,42 @@ impl Engine { autosave, stats, health: Mutex::default(), - mix: program, + mix: program_mix, output, + recorder, room, - _origin: origin, + program, + _origin: subscribe, stopped: AtomicBool::new(false), - program: Mutex::default(), + threads: Mutex::default(), }); - let (video, audio, clock, publisher) = match publish { - Some(Publish { - renditions, - rendition, - audio, - clock, - health, - sources: _, - }) => (Some((renditions, rendition)), Some(audio), Some(clock), Some(health)), - None => (None, None, None, None), - }; + let Publish { + renditions, + rendition, + audio, + clock, + health, + sources: _, + } = publish; let start = Instant::now(); // Where the program clock starts on the catalog's, so the wall time the // catalog advertises for a frame is when the program drew it. - let epoch = match clock { - Some(clock) => Duration::from(clock.capture(start)?), - None => Duration::ZERO, - }; + let epoch = Duration::from(clock.capture(start)?); let program = engine.clone(); let thread = clocked("program".into(), async move || { - program.run_program(start, epoch, video).await + program.run_program(start, epoch, renditions, rendition).await })?; - *engine.program.lock().unwrap() = Some(thread); + engine.threads.lock().unwrap().push(thread); let mix = engine.clone(); let config = bus.config().clone(); - clocked("audio".into(), async move || { - mix.run_audio(start, epoch, config, mixer, audio).await + 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(publisher))); + tokio::spawn(engine.clone().grade(health::Sampler::new(health))); // Last, once nothing can fail: a start that fails is retried from setup, // and a follower already running would keep this engine and its // publication alive beside the retry's. @@ -571,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.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"); + } } } @@ -648,21 +684,28 @@ impl Engine { } /// 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. /// - /// A new canvas size or frame rate ends the rendition of `video` it was - /// drawn at and opens another, since a track never changes what it - /// carries. Drawing outlives the encoder: a failed output stops copying the - /// program, and the preview and program keep moving. + /// A new canvas size or frame rate ends `rendition`, the one it was drawn + /// at, and opens another in `renditions`, since a track never changes what + /// it carries. Drawing outlives the encoder: a failed output stops copying + /// the program, and the preview and program keep moving. /// /// Runs until the studio [stops](Self::stop), and returns once every /// rendition's encode thread has finished its track. - async fn run_program(&self, start: Instant, epoch: Duration, video: Option<(Renditions, Rendition)>) { - let (renditions, mut rendition) = video.unzip(); + async fn run_program( + &self, + start: Instant, + epoch: Duration, + renditions: Renditions, + mut rendition: Option, + ) { let mut canvas = self.state.lock().unwrap().compositor.canvas(); let mut clock = program_clock(start, canvas); - let on_demand = self.output.as_ref().is_some_and(|output| output.on_demand); + // Live on a relay and not on demand; otherwise a recording or a viewer + // asks for the video. + let always = self.output.as_ref().is_some_and(|output| !output.on_demand); // One probe at a time, with the canvas it is for, since a GPU caps its // encode sessions. let mut probing: Option<(Canvas, std::thread::JoinHandle>)> = None; @@ -720,32 +763,30 @@ impl Engine { ending.extend(rendition.take().map(Rendition::end)); self.stats.retire_encoder(); } - if let Some(renditions) = &renditions { - // Probed beside the program clock, so drawing carries on. - if let Some((probed, probe)) = probing.take_if(|(_, probe)| probe.is_finished()) { - let probe = probe - .join() - .unwrap_or_else(|_| Err(anyhow::anyhow!("the encoder probe panicked"))); - // One for a canvas since left is dropped unlisted. - if probed == canvas { - settled = true; - match probe.and_then(|probe| renditions.start(probe)) { - Ok((started, _)) => { - started.adopted(&self.stats); - rendition = Some(started); - } - Err(err) => self.stats.fail_video(format!("{err:#}")), + // Probed beside the program clock, so drawing carries on. + if let Some((probed, probe)) = probing.take_if(|(_, probe)| probe.is_finished()) { + let probe = probe + .join() + .unwrap_or_else(|_| Err(anyhow::anyhow!("the encoder probe panicked"))); + // One for a canvas since left is dropped unlisted. + if probed == canvas { + settled = true; + match probe.and_then(|probe| renditions.start(probe)) { + Ok((started, _)) => { + started.adopted(&self.stats); + rendition = Some(started); } + Err(err) => self.stats.fail_video(format!("{err:#}")), } } - if probing.is_none() && !settled { - let ending = std::mem::take(&mut ending); - match Probe::spawn(renditions.gpu.clone(), canvas, ending) { - Ok(probe) => probing = Some((canvas, probe)), - Err(err) => { - settled = true; - self.stats.fail_video(format!("cannot start an encoder probe: {err}")); - } + } + if probing.is_none() && !settled { + let ending = std::mem::take(&mut ending); + match Probe::spawn(renditions.gpu.clone(), canvas, ending) { + Ok(probe) => probing = Some((canvas, probe)), + Err(err) => { + settled = true; + self.stats.fail_video(format!("cannot start an encoder probe: {err}")); } } } @@ -754,7 +795,7 @@ impl Engine { .as_ref() .is_some_and(|rendition| rendition.watched.load(Ordering::Relaxed)); self.stats.watched.store(watched, Ordering::Relaxed); - let wanted = rendition.is_some() && (!on_demand || watched); + let wanted = rendition.is_some() && (always || watched); self.stats.encoding.store(wanted, Ordering::Relaxed); let Some(current) = &mut rendition else { continue }; let was = std::mem::replace(&mut current.encoding, wanted); @@ -793,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, @@ -810,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; @@ -837,72 +887,97 @@ impl Engine { } } -/// The encoders behind an [`Output`], handed to their threads. +/// The relay connections the program goes live on, the preferred first, and +/// the origin their viewers' reports and the relays' stats arrive on. +struct Live { + connections: Vec, + links: Vec, + reports: moq_net::origin::Consumer, +} + +/// The encoders behind the program's broadcast, handed to their threads. struct Publish { /// Where the program video is published, a rendition at a time. renditions: Renditions, - /// The rendition of the canvas the studio started with. - rendition: Rendition, + /// The rendition of the canvas the studio started with, unless the canvas + /// could not be encoded without a relay. + rendition: Option, audio: moq_audio::encode::Producer, /// 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 under one instance, which also carries each -/// local source's own track, and connect to every relay, each also carrying -/// `subscribe`'s broadcast sources. +/// Announce the program broadcast at `path` on `origin` under one instance, +/// which `live`'s relays publish when there are any. The broadcast also +/// carries each local source's own track. async fn publish( gpu: &Gpu, stats: &Arc, - relays: Vec, - subscribe: moq_net::origin::Producer, - path: &str, + origin: moq_net::origin::Producer, + path: &moq_net::Path, canvas: 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}`"))?; - let reports = subscribe.consume(); - tracing::info!(broadcast = %path, "publishing"); - let (connections, links) = connect(relays, &origin, &subscribe); - - let publication = origin.create_broadcast(&path).context("cannot create the broadcast")?; + live: Option<&Live>, +) -> 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 first relay's estimate, which that // connection then probes to fund: the bandwidth the stream is granted. An // allocator spans one connection, so the failover's link is never probed. - let bandwidth = connections[0].allocator(); + let bandwidth = match live { + Some(live) => live.connections[0].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)?; // Probed now, so a machine that cannot encode the canvas fails at startup - // rather than once the program clock starts. - let probe = Probe::new(gpu, canvas).await?; + // rather than once the program clock starts. Without a relay the studio + // still composes, previews, and records audio, and a new canvas tries again. + let probe = match Probe::new(gpu, canvas).await { + Ok(probe) => Some(probe), + Err(err) if live.is_none() => { + stats.fail_video(format!("{err:#}")); + None + } + Err(err) => return Err(err), + }; // Sources take the program's tier: hardware when the program found it, so a // session the GPU refuses fails that source loudly instead of falling to // software. When the program could not use hardware, such as NVENC beside an // integrated GPU it cannot read, a source may still reach it from the CPU. - let kind = match &probe.config.kind { - moq_video::encode::Kind::Named(name) if name == "openh264" => moq_video::encode::Kind::Auto, + let kind = match probe.as_ref().map(|probe| &probe.config.kind) { + Some(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 live.is_none() { + sources = sources.on_demand(); + } // Held until the video and audio are listed, so the first catalog carries - // both and the solicitation. + // both and any solicitation. let reserved = catalog.reserve(); - 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 live { + Some(_) => { + let (echo, viewers) = health::echo(path)?; + 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 renditions = Renditions { gpu: gpu.clone(), broadcast: broadcast.clone(), @@ -910,11 +985,21 @@ async fn publish( stats: stats.clone(), runtime: tokio::runtime::Handle::current(), }; - let (rendition, opened) = renditions.start(probe)?; - // Before the program clock starts the sources, so the program holds its - // encode session even while unwatched, and a source past the GPU's session - // cap is the one that fails. - opened.await.context("the encode thread stopped")??; + let rendition = match probe { + Some(probe) => { + let (rendition, opened) = renditions.start(probe)?; + // Before the program clock starts the sources, so the program holds + // its encode session even while unwatched, and a source past the + // GPU's session cap is the one that fails. + match opened.await.context("the encode thread stopped")? { + Ok(()) => Some(rendition), + // Its encode thread reported why. + Err(_) if live.is_none() => None, + Err(err) => return Err(err), + } + } + None => None, + }; let bus = moq_audio::bus::Config::default(); let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); @@ -935,33 +1020,35 @@ async fn publish( .announce(moq_net::origin::Route::default().with_instance(instance.clone(), 0)) .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(instance)), - catalog: graded, - clock: clock.clone(), - relays: links, - feedback, - watching, - sources: sources.clone(), - }; + let health = live.zip(viewers).map(|(live, viewers)| { + let feedback = health::Feedback::default(); + let watching = health::Watching::default(); + tokio::spawn(log( + "viewer feedback", + feedback.clone().follow(live.reports.clone(), viewers), + )); + tokio::spawn(log( + "relay stats", + watching.clone().follow(live.reports.clone(), path.clone()), + )); + health::Publisher { + observer: hang::health::Observer::new(path.clone(), Some(instance)), + catalog: graded, + clock: clock.clone(), + relays: live.links.clone(), + feedback, + watching, + sources: sources.clone(), + } + }); - let output = Output { - path: path.to_string(), - on_demand, - _connections: connections, - publication, - catalog: listed, - clock: clock.clone(), - }; Ok(( - output, + Program { + origin, + publication, + catalog: listed, + clock: clock.clone(), + }, Publish { renditions, rendition, @@ -1328,6 +1415,144 @@ mod tests { /// Bounds each step, so a stall fails rather than hangs. const TIMEOUT: Duration = Duration::from_secs(30); + /// How the tests start a studio, live on `relays`. + fn launch(gpu: Gpu, relays: Vec, recordings: &Path) -> Launch { + Launch { + gpu, + relays, + broadcast: "studio".into(), + on_demand: false, + recordings: recordings.to_path_buf(), + room: None, + } + } + + /// A small canvas filled red. + fn red_canvas() -> autosave::Loaded { + let mut collection = Collection::new(); + collection.canvas = moq_scene::Canvas { + width: 320, + height: 180, + framerate: 30, + }; + collection.sources.insert( + "red".into(), + moq_scene::Source::Color { + color: moq_scene::Rgba([255, 0, 0, 255]), + }, + ); + collection.scenes[0] + .items + .push(moq_scene::Item::new("red", moq_scene::Rect::fill(&collection.canvas))); + autosave::Loaded { + collection, + history: History::new(), + dropped: 0, + } + } + + /// Wait until `frames` program frames were encoded. + async fn encoded(engine: &Engine, frames: u64) { + kio::clock::timeout(TIMEOUT, async { + while engine.stats.encoded.load(Ordering::Relaxed) < frames { + kio::clock::sleep(Duration::from_millis(100)).await; + } + }) + .await + .expect("the recording encodes the program"); + } + + /// Finish the recording, and return the handler of every track in each + /// file, by file. + async fn finished(recorder: &record::Recorder) -> Vec> { + 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"); + + 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. A new canvas goes on in another + /// file, holding the new rendition's video. + #[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(launch(gpu, Vec::new(), dir.path()), autosave, red_canvas()) + .await + .expect("the studio starts"); + assert!(engine.output.is_none()); + assert_eq!(engine.stats.encoded.load(Ordering::Relaxed), 0); + + engine.recorder.start(true).expect("recording starts"); + // Long enough for the first group to close, so the video has a fragment. + encoded(&engine, 90).await; + { + let mut state = engine.state.lock().unwrap(); + let mut draft = state.collection().draft("Resize the canvas"); + let canvas = moq_scene::Canvas { + width: 160, + height: 90, + framerate: 15, + }; + draft.set_canvas(canvas).unwrap(); + state.commit(draft, None).unwrap(); + } + encoded(&engine, 90 + 45).await; + let files = finished(&engine.recorder).await; + assert_eq!(files, [["soun", "vide"], ["soun", "vide"]]); + 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 /// not report that encoder over the next rendition's. #[test] @@ -1434,39 +1659,13 @@ mod tests { let client = client.init(Default::default()).expect("client"); let dir = tempfile::tempdir().unwrap(); - let mut collection = Collection::new(); - collection.canvas = moq_scene::Canvas { - width: 320, - height: 180, - framerate: 30, - }; - collection.sources.insert( - "red".into(), - moq_scene::Source::Color { - color: moq_scene::Rgba([255, 0, 0, 255]), - }, - ); - collection.scenes[0] - .items - .push(moq_scene::Item::new("red", moq_scene::Rect::fill(&collection.canvas))); - - let launch = Launch { - gpu: Gpu::new().await.expect("a GPU"), - relays: vec![Relay { - client, - url: fixture.url.clone(), - }], - broadcast: "studio".into(), - on_demand: false, - room: None, - }; + let gpu = Gpu::new().await.expect("a GPU"); let (autosave, _) = Autosave::open(dir.path().join("show.toml")).unwrap(); - let loaded = autosave::Loaded { - collection, - history: History::new(), - dropped: 0, - }; - let engine = Engine::start(launch, autosave, loaded) + let relays = vec![Relay { + client, + url: fixture.url.clone(), + }]; + let engine = Engine::start(launch(gpu, relays, dir.path()), autosave, red_canvas()) .await .expect("the studio starts"); assert!( @@ -1650,6 +1849,14 @@ mod tests { .await .expect("health grades the program's tracks"); + // A recording beside the live program reads the encoder viewers do. + let frames = engine.stats.encoded.load(Ordering::Relaxed); + engine.recorder.start(true).expect("recording starts"); + encoded(&engine, frames + 60).await; + let files = finished(&engine.recorder).await; + assert_eq!(files, [["soun", "vide"]]); + assert!(engine.stats.error().is_none()); + // The program's recent past saves as a clip, from the groups it holds. let target = engine.target(&crate::replay::Of::Program).expect("the program's past"); let take = moq_scene::replay::Take::clip(&target, Duration::from_secs(1)).expect("a take"); diff --git a/rs/moq-studio/src/health.rs b/rs/moq-studio/src/health.rs index 55b1235..92e4d5b 100644 --- a/rs/moq-studio/src/health.rs +++ b/rs/moq-studio/src/health.rs @@ -216,6 +216,7 @@ pub struct Publisher { } /// The connection to one relay. +#[derive(Clone)] pub struct Link { /// Its scheme, host, and port. pub endpoint: String, diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index d7500e8..c2837d8 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::process::ExitCode; use std::sync::{Arc, OnceLock}; @@ -16,6 +16,7 @@ mod engine; mod guests; mod health; mod interact; +mod record; mod replay; mod setup; mod snap; @@ -36,12 +37,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, + /// Guests who join from a web page, appearing as sources. #[usage(flatten)] pub guests: guests::Config, @@ -139,6 +145,15 @@ fn main() -> anyhow::Result { let _runtime = runtime.enter(); let relays = config.relays()?; let gpu = pollster::block_on(engine::Gpu::new())?; + 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 room = config.guests.load(config.client.url.as_ref())?.map(Arc::new); if let Some(room) = &room { let renewal = guests::Renewal::connect(room.clone(), &config.client, config.quic.clone())?; @@ -149,6 +164,7 @@ fn main() -> anyhow::Result { relays, broadcast: config.broadcast.clone(), on_demand: config.on_demand, + recordings, room, }; let exists = config @@ -174,9 +190,14 @@ fn main() -> anyhow::Result { let Some(engine) = started.get() else { return ran.map(|()| ExitCode::SUCCESS); }; + // 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 new file mode 100644 index 0000000..035b1c7 --- /dev/null +++ b/rs/moq-studio/src/record.rs @@ -0,0 +1,1341 @@ +//! 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::{BTreeMap, BTreeSet}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::task::Poll; +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, 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 { + 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>, + /// Set by [`Recorder::close`]. + closing: kio::Producer, + 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 = 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)?; + 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 = kio::Producer::new(false); + let (finished, done) = watch::channel(false); + + let run = Run { + source: self.source.clone(), + dir: self.dir.clone(), + stem, + files: files.clone(), + stop: stop.clone(), + closing: closing.consume(), + }; + 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, + closing, + 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. + #[cfg(test)] + pub async fn finish(&self) { + 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() { + // Fails only once the recording has finished. + if let Ok(mut closing) = session.closing.write() { + *closing = true; + } + } + } + + /// 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; + } + + /// 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>, + /// The recording reads the program to its end; see [`Recorder::close`]. + closing: kio::Consumer, +} + +/// A recording's file writers, each ending with the file it wrote. +type Writers = tokio::task::JoinSet; + +/// A file that ended. +struct Ended { + /// The source whose file it was, or `None` for the program's. + source: Option, + /// Its 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. + async fn record(self, program: Arc, file: tokio::fs::File, sources: bool) { + 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:#}"), "a recording's writer failed"); + self.stop.send_replace(true); + } + // 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 + /// `sources`, give each source that publishes video or audio a file of its + /// own while it is listed, until asked to stop or the program ends. + /// + /// A source removed while recording finishes its file; one added, or added + /// back, starts another, as does one whose tracks were replaced. One whose + /// 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) => match catalog { + Ok(catalog) => catalog, + Err(_) => return Ok(()), + }, + () = stopped(&mut asked) => return Ok(()), + }; + 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 = BTreeMap::new(); + // Sources whose file ended, or could not be created, while their listing + // stayed, held back until they leave the catalog. + let mut ended = BTreeSet::new(); + loop { + tokio::select! { + snapshot = catalog.next() => { + let Ok(Some(snapshot)) = snapshot else { return Ok(()) }; + if !sources { + continue; + } + 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)); + for (name, gone) in &mut writing { + *gone |= !listed.contains(name); + } + } + joined = writers.join_next() => match joined.context("no writer left")?? { + // A listing that went away and came back, or whose tracks were + // replaced, is recorded afresh. + Ended { source: Some(name), replaced } => { + let gone = writing.remove(&name).unwrap_or_default(); + if !gone && !replaced && listed.contains(&name) { + ended.insert(name); + } + } + // 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.read() => { + if !self.file(None, writers) { + self.stop.send_replace(true); + return Ok(()); + } + } + // The program's writer ended, so the recording is over. + Ended { source: None, .. } => { + self.stop.send_replace(true); + return Ok(()); + } + }, + () = stopped(&mut asked) => return Ok(()), + } + if *asked.borrow() || *self.closing.read() { + continue; + } + let start: Vec = listed + .iter() + .filter(|name| !writing.contains_key(*name) && !ended.contains(*name)) + .cloned() + .collect(); + for name in start { + if self.file(Some(&name), writers) { + writing.insert(name, false); + } else { + ended.insert(name); + } + } + } + } + + /// Start recording the program, or the source named, into a new file, + /// returning whether it started rather than failing to create the file. + fn file(&self, source: Option<&str>, writers: &mut Writers) -> bool { + let stem = match source { + None => self.stem.clone(), + Some(name) => source_stem(&self.stem, name), + }; + 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 false; + } + }; + let entry = Arc::new(File::new(path)); + self.files.lock().unwrap().push(entry.clone()); + self.spawn( + source.map(str::to_string), + entry, + tokio::fs::File::from_std(file), + writers, + ); + true + } + + /// Write the program, or the source named, into `file`. + 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, closing).await; + entry.finish(finished); + Ended { source, replaced } + }); + } +} + +/// 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(); + +/// 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 + SOURCE + 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: u32 = 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 = number.checked_add(1).context("every numbered name is taken")?; + } + Err(err) => return Err(err).with_context(|| format!("cannot create {}", path.display())), + } + } +} + +/// Record the program's own tracks from `source`, or with `name` that +/// source's, into `file` until they leave the catalog or are replaced, or +/// `stop` is set, then finish the file. Also returns whether they were +/// replaced. +async fn write( + source: moq_mux::Source, + name: Option<&str>, + file: tokio::fs::File, + entry: &File, + mut stop: watch::Receiver, + 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), + }; + 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.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)) +} + +/// 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 = 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 { + 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 { + // 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 + // 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) +} + +/// The program's own tracks in the studio's catalog, or one source's catalog +/// nested in it, which ends once the source leaves it or the tracks are +/// replaced. +/// +/// A new canvas size or frame rate, or a capture that changes size or color +/// space, moves the video 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, + /// The source, or `None` for the program. + name: Option, + /// 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, + /// Set by [`Recorder::close`]: the program's listing then ends once the + /// tracks it recorded have all ended, adding and replacing none. + closing: kio::Consumer, +} + +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)); + } + // Polled for the program's listing, so closing wakes it. + let closing = self.name.is_none() + && matches!( + self.closing.poll(waiter, |closing| if **closing { + Poll::Ready(()) + } else { + Poll::Pending + }), + Poll::Ready(Ok(())) + ); + loop { + 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 { + None => Some(studio.media()), + Some(name) => studio.ext.studio.sources.get(name).cloned(), + }; + let Some(listing) = listing else { + if self.last.is_some() { + return Poll::Ready(Ok(None)); + } + continue; + }; + 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. + 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) + }; + 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); + 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))); + } + } + self.last = Some(listing.clone()); + return Poll::Ready(Ok(Some(listing))); + } + } +} + +/// 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; + 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, + /// `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)>, + /// 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 = Some(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) { + // 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(); + } + + /// 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(); + } + + /// Move the program onto a new track, as a new canvas size or frame rate + /// moves its video. + fn replace_program(&mut self) { + 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 = self.program.replace(tone).unwrap(); + old_tone.audio.finish().unwrap(); + drop(old_tone); + self.catalog + .mutate(|root| root.audio.renditions.retain(|track, _| !old.contains_key(track))) + .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 + .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 { + if let Some(program) = &mut self.program { + 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, from [`MAX_DELAY`] before it + /// started. + #[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_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; + 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, MAX_DELAY + 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, 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(); + 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; + // 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(); + 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", " mic_1 (2).mp4"]); + for ((_, duration), expected) in files.iter().zip([4, 2, 3, 1]) { + assert!(about(*duration, Duration::from_secs(expected)), "{files:?}"); + } + } + + /// 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 program moved onto a new track, as a new canvas moves its video, goes + /// on in another file, with or without the sources. + #[tokio::test(start_paused = true)] + async fn a_replaced_program_goes_on_in_another_file() { + let dir = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + let recorder = studio.recorder(dir.path()); + + recorder.start(false).unwrap(); + studio.play(Duration::from_secs(1)).await; + studio.replace_program(); + 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", " (2).mp4"]); + for ((_, duration), expected) in files.iter().zip([1, 2]) { + assert!(about(*duration, Duration::from_secs(expected)), "{files:?}"); + } + } + + /// 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:?}"); + } + + /// 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 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. 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()); + 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(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 + /// 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)] + #[tokio::test(start_paused = true)] + async fn a_source_that_failed_to_start_records_once_added_back() { + use std::os::unix::fs::PermissionsExt; + + let dir = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + let recorder = studio.recorder(dir.path()); + recorder.start(true).unwrap(); + studio.play(Duration::from_millis(500)).await; + + let mode = |mode| std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(mode)).unwrap(); + mode(0o555); + studio.add("mic"); + studio.play(Duration::from_secs(1)).await; + studio.remove("mic"); + studio.play(Duration::from_millis(500)).await; + mode(0o755); + // Fixed, it still waits to be listed again. + studio.play(Duration::from_millis(500)).await; + assert_eq!(recorder.status().unwrap().files.len(), 2); + studio.add("mic"); + studio.play(Duration::from_secs(1)).await; + recorder.finish().await; + + let status = recorder.status().unwrap(); + let outcomes: Vec<_> = status + .files + .iter() + .map(|file| { + let name = file.path.file_name().unwrap().to_string_lossy(); + (name.rsplit_once('Z').unwrap().1.to_string(), file.finished.clone()) + }) + .collect(); + let [program, failed, added] = outcomes.as_slice() else { + panic!("{outcomes:?}"); + }; + assert_eq!(program, &(".mp4".to_string(), Some(Finished::Saved))); + assert_eq!(failed.0, " mic.mp4"); + assert!( + matches!(&failed.1, Some(Finished::Failed(reason)) if reason.contains("cannot create")), + "{failed:?}" + ); + assert_eq!(added, &(" mic.mp4".to_string(), Some(Finished::Saved))); + let [(_, duration)] = tracks(&status.files[2].path).try_into().unwrap(); + assert!(about(duration, Duration::from_secs(1)), "{duration:?}"); + } + + /// 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 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(); + 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 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()); + // 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 { + 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 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 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(); + 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] + 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/replay.rs b/rs/moq-studio/src/replay.rs index b639cc6..c87dbe8 100644 --- a/rs/moq-studio/src/replay.rs +++ b/rs/moq-studio/src/replay.rs @@ -45,13 +45,14 @@ impl Engine { /// What a replay of `of` copies, ending now. /// /// The program and the local sources have a past only while they are - /// published, since nothing else encodes them, and a broadcast source once - /// it shows a picture. + /// published, since without a relay they encode only while recorded, and + /// a broadcast source once it shows a picture. pub fn target(&self, of: &Of) -> anyhow::Result { let output = || { self.output .as_ref() - .context("nothing is encoded without --connect, so there is no past to replay") + .map(|_| &self.program) + .context("without --connect, nothing holds a past to replay") }; let name = match of { Of::Program => return Ok(output()?.program()), diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index a10b375..74292d4 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -21,6 +21,7 @@ use crate::devices::{self, Listing, Pick}; use crate::engine::{Engine, State}; use crate::health::{self, Verdict}; use crate::interact::Interact; +use crate::record; use crate::replay::{self, Of}; use crate::snap::{self, Grab, Snap}; @@ -111,6 +112,7 @@ impl Studio { new_item: None, add: AddSource::default(), devices: Devices::default(), + record_sources: false, interacting: false, interact: Interact::default(), previewed: false, @@ -216,6 +218,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, /// Whether the preview passes the mouse and keyboard to browser sources. interacting: bool, interact: Interact, @@ -1683,13 +1687,60 @@ 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 = &engine.recorder; + 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(); self.canvas_form(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 { @@ -1698,10 +1749,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(); } @@ -1709,6 +1760,13 @@ impl Panels { ui.label("Relay"); ui.label("none; start with --connect to publish"); ui.end_row(); + ui.label("Encoding"); + ui.label(if watched { + "while recorded; recording now" + } else { + "while recorded; idle now" + }); + ui.end_row(); } } ui.label("Drawn");