diff --git a/Cargo.lock b/Cargo.lock index 68fb70c..36db508 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4272,6 +4272,7 @@ dependencies = [ "mp4-atom", "pollster 1.0.1", "qrcodegen", + "same-file", "serde", "tempfile", "tokio", diff --git a/README.md b/README.md index 7857589..037ab46 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,7 @@ Every panel is a tab you can drag, split, or stack. | Keyframes | Animate the selected item: add, time, ease, and remove its keyframes, and set how far each moves, resizes, crops, and filters it (see [Keyframes](#keyframes)). **Play** plays the preview scene's keyframes again. | | Audio | Add, rename (double-click), and remove mixes, pick the one on program and the one to listen to and on which output, and set each audio source's gain, mute, delay, and the mixes it feeds, with a meter on every mix and source. | | Output | Start and stop recording, pick the mix it records, set the canvas size and frame rate, add, resize, and remove further canvases, and see the broadcast path, whether the program is encoding, the encoder sessions open, and each canvas's renditions' encoder work. | -| Health | See how the stream reaches viewers, relay by relay and track by track, and the cause of each problem. | +| Health | See how the stream reaches viewers, relay by relay and track by track, and the cause of each problem, and which recordings a crash left unfinished were repaired at launch. | | 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. | A new source, from the Sources or Devices panel, is placed over whatever the @@ -746,6 +746,16 @@ Quitting first lets the recording's encoders publish what they still hold, and the program's file keeps it. A file nothing reached before the recording stopped is removed. +While a file is written, a marker `.mp4.unfinished` sits beside it. A +crash, a power loss, or a write that failed leaves the marker, and the next +launch repairs that file without asking: it writes `.repaired.mp4` +beside it, numbered if a file already has that name, a regular MP4 of +everything the file holds up to its last complete fragment, leaves the +original untouched, and removes the marker, so the file is repaired +only once. The repair runs in the background and appears at the top of the +Health panel and in the log. A file another studio is still recording into the +same directory is left alone. + ### Source frames Sources make the opposite trip. A frame already on the GPU is drawn where it diff --git a/quest/m0/README.md b/quest/m0/README.md index 2f80f7b..838bf5b 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -101,7 +101,6 @@ repository. - [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 - [Web studio](/quest/m0/web.md) - the full studio in wasm on WebGPU and WebCodecs, served from this repository - [Virtual camera](/quest/m0/vcam/README.md) - a canvas's program exposed as a system camera on Linux, Windows, and macOS -- [Recover crashed recordings](/quest/m0/recording-repair.md) - a recording a crash left unfinished is repaired beside the original at launch - [Browser sources without an X display](/quest/m0/cef-headless.md) - CEF runs headless, so Wayland sessions and tests need no X server - [Signing identities are available](/quest/m0/signing.md) - the Apple and Windows code-signing credentials packaging and the macOS camera need - [Packaging](/quest/m0/packaging.md) - signed installers for macOS and Windows and a Flatpak for Linux diff --git a/quest/m0/recording-repair.md b/quest/m0/recording-repair.md deleted file mode 100644 index 3a26f8b..0000000 --- a/quest/m0/recording-repair.md +++ /dev/null @@ -1,33 +0,0 @@ -# [S] Recover crashed recordings - -## Goal - -When the studio starts and finds a recording that a crash or power loss left -unfinished, it repairs the file without asking, writes the result as -`.repaired.mp4` beside the original, and reports the repair in Health -and the log. The original stays untouched. - -## Plan - -From [moq-dev/studio#18](https://github.com/moq-dev/studio/pull/18). - -Decided 2026-11-04 with the maintainer (/quest-plan): - -- ✅ **Repair automatically and keep the original.** The repaired copy is - written to a new file, so a repair that goes wrong never costs the only - copy, and no prompt blocks startup. - Rejected: repair in place (risks the only copy), ask first (a prompt at - every launch after a crash). - -Implementation: - -- `moq_mux::container::mp4::repair` rebuilds a file from its fragments. -- The recording directory and file naming are in `rs/moq-studio/src/record.rs`. - Mark a file as in progress with a sidecar (for example `.mp4.part`) - written when it starts and deleted when it finishes. Launch repairs each - file with a sidecar, then deletes the sidecar once `.repaired.mp4` is - written, so the same file is never repaired twice and the original keeps - its name. -- Repair each source's file as well as the program's. -- A test writes a recording, stops before finishing it, and checks that - launch produces a playable `.repaired.mp4` holding the samples written. diff --git a/rs/moq-studio/Cargo.toml b/rs/moq-studio/Cargo.toml index a4c8ebe..b465a38 100644 --- a/rs/moq-studio/Cargo.toml +++ b/rs/moq-studio/Cargo.toml @@ -69,6 +69,9 @@ moq-video = { workspace = true } pollster = { workspace = true } # The join link's QR code. No dependencies of its own. qrcodegen = "1.8" +# Whether a recording's marker is still the file its holder locked, which std +# cannot tell on Windows. +same-file = "1.0.6" # The guests removed from scenes, kept beside the scene file. serde = { workspace = true } tempfile = { workspace = true } diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 48d0d07..37b0b05 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -6,7 +6,9 @@ //! relay serves, and each source's track from the studio's local broadcast, //! which its reading keeps encoding. 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. +//! when the recording stops. A sidecar marks each file while it is written, so +//! the next launch finds one a crash left unfinished and repairs it (see +//! [`repair`]). use std::collections::{BTreeMap, BTreeSet}; use std::path::{Path, PathBuf}; @@ -22,6 +24,9 @@ use moq_mux::container::mp4; use tokio::io::{AsyncSeekExt, AsyncWriteExt}; use tokio::sync::watch; +mod repair; +pub use repair::{Repair, Repaired}; + /// 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 @@ -38,6 +43,8 @@ pub struct Recorder { runtime: tokio::runtime::Handle, /// The recording in progress, or the last one. session: Mutex>, + /// The recordings found unfinished in `dir` at launch. + repairs: repair::Repairs, } struct Session { @@ -58,6 +65,9 @@ struct File { written: AtomicU64, /// `None` while it is being written. finished: Mutex>, + /// Marks the file unfinished until it is a regular MP4 or removed. A file + /// that fails before then keeps it, so the next launch repairs it. + marker: Mutex>, } /// How a file ended. @@ -68,7 +78,8 @@ pub enum Finished { /// 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. + /// it got, as a fragmented MP4 unless it finished, and one left unfinished + /// keeps its marker, so the next launch repairs it. Failed(String), } @@ -93,7 +104,8 @@ pub struct FileStatus { impl Recorder { /// Record `program`, the studio's broadcast, and from `sources`, the local /// one, each source's own track, into `dir`, naming each file after the - /// `scenes` file. + /// `scenes` file, and repair in the background each file in `dir` a crash + /// left unfinished. /// /// Call inside the tokio runtime the recordings write on. pub fn new(program: moq_mux::Source, sources: moq_mux::Source, dir: &Path, scenes: &Path) -> anyhow::Result { @@ -102,12 +114,15 @@ impl Recorder { .context("the scene file has no name")? .to_string_lossy() .into_owned(); + let dir = std::path::absolute(dir)?; + let runtime = tokio::runtime::Handle::try_current().context("not inside a tokio runtime")?; Ok(Self { program, sources, - dir: std::path::absolute(dir)?, + repairs: repair::Repairs::start(&dir, &runtime), + dir, name, - runtime: tokio::runtime::Handle::try_current().context("not inside a tokio runtime")?, + runtime, session: Mutex::new(None), }) } @@ -135,8 +150,8 @@ impl Recorder { 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 (path, file, marker) = create(&self.dir, &stem)?; + let program = Arc::new(File::new(path, Some(marker))); let files: Files = Arc::new(Mutex::new(vec![program.clone()])); let stop = Arc::new(watch::channel(false).0); let closing = kio::Producer::new(false); @@ -209,6 +224,19 @@ impl Recorder { let _ = done.wait_for(|done| *done).await; } + /// Each recording found unfinished at launch, and how its repair went. + pub fn repairs(&self) -> Vec { + self.repairs.list() + } + + /// Wait until every repair started at launch has ended. + #[cfg(test)] + pub async fn repaired(&self) { + while self.repairs().iter().any(|repair| repair.outcome.is_none()) { + kio::clock::sleep(Duration::from_millis(10)).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(); @@ -231,11 +259,19 @@ impl Recorder { } impl File { - fn new(path: PathBuf) -> Self { + fn new(path: PathBuf, marker: Option) -> Self { Self { path, written: AtomicU64::new(0), finished: Mutex::new(None), + marker: Mutex::new(marker), + } + } + + /// The file is a regular MP4, or removed, so nothing is left to repair. + fn settle(&self) { + if let Some(marker) = self.marker.lock().unwrap().take() { + marker.clear(); } } @@ -245,6 +281,10 @@ impl File { Finished::Failed(format!("{err:#}")) }); *self.finished.lock().unwrap() = Some(finished); + // Nothing writes the file any more, so let go of the lock of a marker a + // failure left, keeping the marker, and another studio's launch may + // repair the file while this one still runs. + drop(self.marker.lock().unwrap().take()); } } @@ -385,16 +425,16 @@ impl Run { None => self.stem.clone(), Some(name) => source_stem(&self.stem, name), }; - let (path, file) = match create(&self.dir, &stem) { + let (path, file, marker) = match create(&self.dir, &stem) { Ok(created) => created, Err(err) => { - let entry = File::new(self.dir.join(format!("{stem}.mp4"))); + let entry = File::new(self.dir.join(format!("{stem}.mp4")), None); entry.finish(Err(err)); self.files.lock().unwrap().push(Arc::new(entry)); return false; } }; - let entry = Arc::new(File::new(path)); + let entry = Arc::new(File::new(path, Some(marker))); self.files.lock().unwrap().push(entry.clone()); self.spawn( source.map(str::to_string), @@ -425,8 +465,20 @@ impl Run { /// Windows counts UTF-16 units instead, which are never more. const MAX_NAME: usize = 255; -/// The most [`create`] adds to a stem: the largest number, and the extension. -const SUFFIX: usize = " (4294967295).mp4".len(); +/// The most a recording's files add to a stem: the largest number [`create`] +/// gives, and the longest name a file takes from there, its repair's, which +/// is numbered too. +const SUFFIX: usize = 2 * NUMBER.len() + REPAIRED.len() + ".mp4".len(); + +/// The largest number a taken name is given. +const NUMBER: &str = " (4294967295)"; + +/// What a file's marker adds to its name. Specific to the studio, since a +/// launch removes a marker whose recording is gone. +const MARKER: &str = ".unfinished"; + +/// What a repaired file's stem adds to the recording's. +const REPAIRED: &str = ".repaired"; /// 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. @@ -463,26 +515,111 @@ fn truncate(text: &str, bytes: usize) -> &str { &text[..end] } +/// The name of the `number`th file named `stem`. +fn numbered(stem: &str, number: u32) -> String { + match number { + 1 => format!("{stem}.mp4"), + _ => format!("{stem} ({number}).mp4"), + } +} + /// 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)> { +/// one already there, and mark it unfinished before it exists. +fn create(dir: &Path, stem: &str) -> anyhow::Result<(PathBuf, std::fs::File, Marker)> { 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")?; + let path = dir.join(numbered(stem, number)); + if let Some(marker) = Marker::create(&path)? { + match std::fs::File::create_new(&path) { + Ok(file) => return Ok((path, file, marker)), + Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => marker.clear(), + Err(err) => { + marker.clear(); + return Err(err).with_context(|| format!("cannot create {}", path.display())); + } } + } + number = number.checked_add(1).context("every numbered name is taken")?; + } +} + +/// The sidecar `.mp4.unfinished` beside a file being recorded, until the file +/// is a regular MP4 or removed. A crash leaves it, which tells the next launch +/// to repair the file. Its lock, held while recording, tells a launch apart +/// from another studio still recording into the same directory. +struct Marker { + path: PathBuf, + /// Holds the lock; closing it releases the lock. + file: std::fs::File, +} + +impl Marker { + /// Mark the recording at `path`, created next, unfinished, or `None` when + /// the name is taken: a marker is already there, or a launch claimed this + /// one before it was locked. + /// + /// Locked before the recording exists, so a launch that claims it first + /// finds no recording and removes it rather than repairing a live one. + fn create(path: &Path) -> anyhow::Result> { + let path = marker(path); + let file = match std::fs::File::create_new(&path) { + Ok(file) => file, + Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => return Ok(None), Err(err) => return Err(err).with_context(|| format!("cannot create {}", path.display())), + }; + match file.try_lock() { + Ok(()) => {} + // A launch holds it, and removes it once it finds no recording. + Err(std::fs::TryLockError::WouldBlock) => return Ok(None), + // A file system without locks still marks the file; only another studio + // recording into the same directory as it launches goes unnoticed. + Err(std::fs::TryLockError::Error(err)) => { + tracing::warn!(path = %path.display(), error = %err, "cannot lock a recording's marker"); + } + } + // A launch that held it before it was locked removed it, and another + // recording may have marked the same name since. + if !Self::named(&path, &file)? { + return Ok(None); + } + Ok(Some(Self { path, file })) + } + + /// Whether `path` still names `file`, which is locked, rather than a marker + /// created there after it was removed. Whoever removes a marker holds its + /// lock first, so the answer stands while the lock is held. Compares file + /// identities rather than reading `path`, which Windows refuses while the + /// lock is held. + fn named(path: &Path, file: &std::fs::File) -> anyhow::Result { + let held = file.try_clone().and_then(same_file::Handle::from_file); + let held = held.with_context(|| format!("cannot identify {}", path.display()))?; + match same_file::Handle::from_path(path) { + Ok(named) => Ok(named == held), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(false), + Err(err) => Err(err).with_context(|| format!("cannot identify {}", path.display())), + } + } + + /// Remove the marker, and only then let go of its lock, so a launch that + /// takes the lock next finds it gone. + fn clear(self) { + if let Err(err) = std::fs::remove_file(&self.path) + && err.kind() != std::io::ErrorKind::NotFound + { + // The next launch finds the file finished or gone, and removes it. + tracing::warn!(path = %self.path.display(), error = %err, "cannot remove a recording's marker"); } + drop(self.file); } } +/// The marker of the recording at `path`. +fn marker(path: &Path) -> PathBuf { + let mut name = path.as_os_str().to_owned(); + name.push(MARKER); + name.into() +} + /// 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 @@ -573,6 +710,7 @@ async fn record( file.seek(std::io::SeekFrom::Start(finish.mdat_offset)).await?; file.write_all(&finish.mdat).await?; file.sync_all().await?; + entry.settle(); entry .written .store(finish.moov_offset + finish.moov.len() as u64, Ordering::Relaxed); @@ -590,6 +728,7 @@ async fn stopped(stop: &mut watch::Receiver) { async fn remove(file: tokio::fs::File, entry: &File) -> anyhow::Result { drop(file); tokio::fs::remove_file(&entry.path).await?; + entry.settle(); Ok(Finished::Empty) } @@ -1339,6 +1478,196 @@ mod tests { assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0); } + /// The names in `dir`, sorted. + fn names(dir: &Path) -> Vec { + let mut names: Vec<_> = std::fs::read_dir(dir) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + names.sort(); + names + } + + /// A recording a crash left unfinished, the program's file and each + /// source's, is repaired at the next launch into a regular MP4 beside it + /// holding what was written, leaving the original as it was, and only once. + /// One that finished before the crash removed its marker is left alone. + #[tokio::test(start_paused = true)] + async fn launch_repairs_a_crashed_recording() { + let dir = tempfile::tempdir().unwrap(); + let crashed = tempfile::tempdir().unwrap(); + let mut studio = Studio::new(); + studio.add("mic"); + // Its file is named as the mic's repair would be. + studio.add("mic.repaired"); + let recorder = studio.recorder(dir.path()); + recorder.start(true).unwrap(); + studio.play(Duration::from_secs(1)).await; + + // What a crash leaves: each file as written so far, and its marker. + let names_while_recording = names(dir.path()); + assert_eq!(names_while_recording.len(), 6, "{names_while_recording:?}"); + for name in &names_while_recording { + std::fs::copy(dir.path().join(name), crashed.path().join(name)).unwrap(); + } + recorder.finish().await; + assert!( + names(dir.path()).iter().all(|name| name.ends_with(".mp4")), + "a finished recording leaves no marker: {:?}", + names(dir.path()) + ); + // A file finished just before a crash, whose marker stayed. + let program = &recorder.status().unwrap().files[0].path; + std::fs::copy(program, crashed.path().join("finished.mp4")).unwrap(); + std::fs::write(crashed.path().join("finished.mp4.unfinished"), b"").unwrap(); + + let originals: Vec<_> = names(crashed.path()) + .into_iter() + .map(|name| { + let data = std::fs::read(crashed.path().join(&name)).unwrap(); + (name, data) + }) + .collect(); + // A read-only original repairs too. + #[cfg(unix)] + for name in names(crashed.path()).iter().filter(|name| name.ends_with(".mp4")) { + use std::os::unix::fs::PermissionsExt; + let permissions = std::fs::Permissions::from_mode(0o444); + std::fs::set_permissions(crashed.path().join(name), permissions).unwrap(); + } + let launch = studio.recorder(crashed.path()); + kio::clock::timeout(Duration::from_secs(5), launch.repaired()) + .await + .expect("the repairs end"); + + // The name after the timestamp, or all of it. + let end = |path: &Path| { + let name = path.file_name().unwrap().to_string_lossy().into_owned(); + name.rsplit_once('Z').map_or(name.clone(), |(_, end)| end.to_string()) + }; + let repairs = launch.repairs(); + let outcomes: Vec<_> = repairs + .iter() + .map(|repair| { + let outcome = match &repair.outcome { + Some(Repaired::Saved(repaired)) => Some(end(repaired)), + Some(Repaired::Finished) => None, + outcome => panic!("{}: {outcome:?}", repair.path.display()), + }; + (end(&repair.path), outcome) + }) + .collect(); + // In name order, where a space comes before a dot. The repair of the + // mic's file is numbered rather than replacing the other source's. + let expected = [ + ("finished.mp4", None), + (" mic.mp4", Some(" mic.repaired (2).mp4")), + (" mic.repaired.mp4", Some(" mic.repaired.repaired.mp4")), + (".mp4", Some(".repaired.mp4")), + ] + .map(|(name, repaired)| (name.to_string(), repaired.map(str::to_string))); + assert_eq!(outcomes, expected); + for repair in &repairs[1..] { + let Some(Repaired::Saved(repaired)) = &repair.outcome else { + unreachable!(); + }; + let [(handler, duration)] = tracks(repaired).try_into().unwrap(); + assert_eq!(handler, "soun"); + assert!( + about(duration, Duration::from_secs(1)), + "{}: {duration:?}", + repaired.display() + ); + } + + // Each original is as it was, and only the repairs were added. + for (name, data) in &originals { + let path = crashed.path().join(name); + if name.ends_with(MARKER) { + assert!(!path.exists(), "{name} is removed once repaired"); + } else { + assert!(&std::fs::read(&path).unwrap() == data, "{name} is left as it was"); + } + } + assert_eq!(names(crashed.path()).len(), originals.len() - 4 + 3); + + // The next launch has nothing to repair. + let again = studio.recorder(crashed.path()); + assert!(again.repairs().is_empty()); + } + + /// What else a launch may find: a file nothing reached and one that is no + /// recording each give up their marker, as does a marker whose file is + /// gone, and a file another studio is still recording is left alone. + #[tokio::test(start_paused = true)] + async fn launch_leaves_what_it_cannot_repair() { + let dir = tempfile::tempdir().unwrap(); + let studio = Studio::new(); + std::fs::write(dir.path().join("empty.mp4"), b"").unwrap(); + std::fs::write(dir.path().join("empty.mp4.unfinished"), b"").unwrap(); + std::fs::write(dir.path().join("junk.mp4"), b"not a recording").unwrap(); + std::fs::write(dir.path().join("junk.mp4.unfinished"), b"").unwrap(); + std::fs::write(dir.path().join("gone.mp4.unfinished"), b"").unwrap(); + let (_, _file, _marker) = create(dir.path(), "recording").unwrap(); + + let launch = studio.recorder(dir.path()); + kio::clock::timeout(Duration::from_secs(5), launch.repaired()) + .await + .expect("the repairs end"); + let outcomes: Vec<_> = launch + .repairs() + .into_iter() + .map(|repair| { + ( + repair.path.file_name().unwrap().to_string_lossy().into_owned(), + repair.outcome, + ) + }) + .collect(); + let [(empty, Some(Repaired::Empty)), (junk, Some(Repaired::Failed(reason)))] = outcomes.as_slice() else { + panic!("{outcomes:?}"); + }; + assert_eq!([empty.as_str(), junk], ["empty.mp4", "junk.mp4"]); + assert!(reason.contains("not a recording the studio can repair"), "{reason}"); + assert_eq!( + names(dir.path()), + ["empty.mp4", "junk.mp4", "recording.mp4", "recording.mp4.unfinished"] + ); + } + + /// A file the launch cannot read keeps its marker, so the next launch tries + /// again, unlike one it reads and finds no recording in. + #[cfg(unix)] + #[tokio::test(start_paused = true)] + async fn a_repair_that_cannot_read_tries_again() { + use std::os::unix::fs::PermissionsExt; + + let dir = tempfile::tempdir().unwrap(); + let studio = Studio::new(); + let path = dir.path().join("locked.mp4"); + std::fs::write(&path, b"not a recording").unwrap(); + std::fs::write(dir.path().join("locked.mp4.unfinished"), b"").unwrap(); + let launch = async |mode| { + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode)).unwrap(); + let launch = studio.recorder(dir.path()); + kio::clock::timeout(Duration::from_secs(5), launch.repaired()) + .await + .expect("the repairs end"); + let [repair] = launch.repairs().try_into().unwrap(); + let Some(Repaired::Failed(reason)) = repair.outcome else { + panic!("{:?}", repair.outcome); + }; + (reason, names(dir.path())) + }; + + let (reason, names) = launch(0o000).await; + assert!(reason.contains("cannot open"), "{reason}"); + assert_eq!(names, ["locked.mp4", "locked.mp4.unfinished"]); + let (reason, names) = launch(0o644).await; + assert!(reason.contains("not a recording the studio can repair"), "{reason}"); + assert_eq!(names, ["locked.mp4"]); + } + /// A long scene file or source name is cut, between characters, so every /// file name fits, numbered ones included. #[test] @@ -1353,9 +1682,13 @@ mod tests { 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"); + let (path, ..) = create(dir.path(), stem).unwrap(); + // So do its marker and its repair, numbered too. + let repaired = dir.path().join(numbered(&repair::stem(&path).unwrap(), u32::MAX)); + for path in [marker(&path), repaired, path] { + let name = path.file_name().unwrap().len(); + assert!(name <= MAX_NAME, "{name} bytes: {}", path.display()); + } } } } @@ -1393,14 +1726,46 @@ mod tests { #[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(); + 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")); } + /// A marker already there, such as one a launch holds while it removes it, + /// takes its name too, so a new recording is never marked by a marker that + /// someone else may remove. + #[test] + fn a_taken_marker_is_numbered() { + let dir = tempfile::tempdir().unwrap(); + let held = std::fs::File::create(dir.path().join("show.mp4.unfinished")).unwrap(); + held.try_lock().unwrap(); + let (path, _file, created) = create(dir.path(), "show").unwrap(); + assert_eq!(path, dir.path().join("show (2).mp4")); + assert_eq!(created.path, marker(&path)); + assert_eq!( + names(dir.path()), + ["show (2).mp4", "show (2).mp4.unfinished", "show.mp4.unfinished"] + ); + } + + /// A marker removed while its holder was not looking, and replaced by + /// another recording's at the same name, is no longer its holder's. + // Windows keeps a removed name taken while a handle to it is open. + #[cfg(unix)] + #[test] + fn a_replaced_marker_is_not_its_holders() { + let dir = tempfile::tempdir().unwrap(); + let recording = dir.path().join("show.mp4"); + let Marker { path, file } = Marker::create(&recording).unwrap().unwrap(); + assert!(Marker::named(&path, &file).unwrap()); + std::fs::remove_file(&path).unwrap(); + let _other = Marker::create(&recording).unwrap().unwrap(); + assert!(!Marker::named(&path, &file).unwrap()); + } + /// Starting into a directory that does not exist fails at once. #[tokio::test(start_paused = true)] async fn a_missing_directory_fails_to_start() { diff --git a/rs/moq-studio/src/record/repair.rs b/rs/moq-studio/src/record/repair.rs new file mode 100644 index 0000000..46d45f4 --- /dev/null +++ b/rs/moq-studio/src/record/repair.rs @@ -0,0 +1,309 @@ +//! Finishing the recordings a crash or power loss left unfinished. +//! +//! Each file a recording writes has a sidecar marker beside it while it is +//! written (see [`Marker`](super::Marker)). A marker found at launch whose +//! lock nobody holds names a file that never finished. The launch takes the +//! lock for as long as it repairs the file into a regular MP4 beside it, +//! `.repaired.mp4`, numbered rather than replacing a file already +//! there, leaving the original as it was, so neither a +//! recording nor another studio's launch touches it meanwhile. Removing the +//! marker after the repair keeps the file from being repaired again. + +use std::io::{Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; + +use anyhow::Context; +use moq_mux::container::mp4; + +use super::Marker; + +/// A recording found unfinished at launch. +#[derive(Clone, Debug)] +pub struct Repair { + /// The unfinished file, which the repair leaves as it was. + pub path: PathBuf, + /// `None` while it is being repaired. + pub outcome: Option, +} + +/// How a repair ended. +#[derive(Clone, Debug, PartialEq)] +pub enum Repaired { + /// Written as a regular MP4 at this path. + Saved(PathBuf), + /// Already a regular MP4: the studio stopped just after finishing it. + Finished, + /// Nothing reached it before the studio stopped. + Empty, + /// It holds no recording that can be repaired, or reading it or writing + /// the repair failed. A failure to read or write leaves the marker, so the + /// next launch tries again. + Failed(String), +} + +/// The repairs a launch started, running in the background. +pub struct Repairs { + list: Arc>>, +} + +impl Repairs { + /// Repair each recording in `dir` that a crash left unfinished, on + /// `runtime`'s blocking pool, so a long file never holds up the launch. + /// + /// Only the markers there now count, not those of recordings started later. + pub fn start(dir: &Path, runtime: &tokio::runtime::Handle) -> Self { + let claimed = unfinished(dir); + let list: Vec<_> = claimed + .iter() + .map(|(path, _)| Repair { + path: path.clone(), + outcome: None, + }) + .collect(); + let list = Arc::new(Mutex::new(list)); + if !claimed.is_empty() { + let list = list.clone(); + runtime.spawn_blocking(move || { + for (index, (path, marker)) in claimed.into_iter().enumerate() { + let outcome = repair(&path, marker); + list.lock().unwrap()[index].outcome = Some(outcome); + } + }); + } + Self { list } + } + + /// Every recording found unfinished at launch, in name order. + pub fn list(&self) -> Vec { + self.list.lock().unwrap().clone() + } +} + +/// Each file in `dir` whose marker no running recording or repair holds, in +/// name order, with its marker claimed. A marker whose file is gone is removed. +fn unfinished(dir: &Path) -> Vec<(PathBuf, Marker)> { + let entries = match std::fs::read_dir(dir) { + Ok(entries) => entries, + // Nothing was ever recorded there. + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Vec::new(), + Err(err) => { + tracing::warn!(dir = %dir.display(), error = %err, "cannot look for unfinished recordings"); + return Vec::new(); + } + }; + let mut claimed: Vec<_> = entries + .filter_map(|entry| { + let path = entry.ok()?.path(); + let name = path.file_name()?.to_str()?.strip_suffix(super::MARKER)?; + if !name.ends_with(".mp4") { + return None; + } + let recording = dir.join(name); + let marker = Marker::claim(path)?; + match recording.try_exists() { + Ok(true) => Some((recording, marker)), + Ok(false) => { + tracing::debug!(marker = %marker.path.display(), "removing the marker of a recording that is gone"); + marker.clear(); + None + } + // Dropping the marker keeps it for the next launch. + Err(err) => { + tracing::warn!(path = %recording.display(), error = %err, "cannot look for an unfinished recording"); + None + } + } + }) + .collect(); + claimed.sort_by(|(a, _), (b, _)| a.cmp(b)); + claimed +} + +impl Marker { + /// Claim the marker at `path` for a repair, unless a recording, such as + /// another studio's into the same directory, or another launch's repair + /// holds it. + fn claim(path: PathBuf) -> Option { + // Writable, as an exclusive lock needs on NFS. + let file = match std::fs::OpenOptions::new().read(true).write(true).open(&path) { + Ok(file) => file, + // Its recording finished meanwhile. + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return None, + Err(err) => { + tracing::warn!(path = %path.display(), error = %err, "cannot open a recording's marker"); + return None; + } + }; + match file.try_lock() { + Ok(()) => {} + Err(std::fs::TryLockError::WouldBlock) => return None, + // A file system without locks: claimed, as the recording could not + // lock it either. + Err(std::fs::TryLockError::Error(err)) => { + tracing::debug!(path = %path.display(), error = %err, "cannot lock a recording's marker"); + } + } + // Whoever held it before removed it, and a new recording may have + // marked the same name since. + match Self::named(&path, &file) { + Ok(true) => {} + Ok(false) => return None, + Err(err) => { + tracing::warn!(error = %format!("{err:#}"), "cannot claim a recording's marker"); + return None; + } + } + Some(Self { path, file }) + } +} + +/// Repair the unfinished recording at `path`, reporting it in the log, and +/// remove its `marker` unless the next launch should try again. +fn repair(path: &Path, marker: Marker) -> Repaired { + let (outcome, again) = write(path); + if !again { + marker.clear(); + } + match &outcome { + Repaired::Saved(repaired) => tracing::warn!( + path = %path.display(), + repaired = %repaired.display(), + "repaired a recording the studio left unfinished" + ), + Repaired::Finished => tracing::info!(path = %path.display(), "an unfinished recording was already finished"), + Repaired::Empty => tracing::info!(path = %path.display(), "an unfinished recording holds nothing"), + Repaired::Failed(reason) => tracing::error!( + path = %path.display(), + error = %reason, + "cannot repair a recording the studio left unfinished" + ), + } + outcome +} + +/// Write the repair of `path` beside it. Also returns whether the next launch +/// should try again: after a failure to read or write rather than a file that +/// holds no recording. +fn write(path: &Path) -> (Repaired, bool) { + if std::fs::metadata(path).is_ok_and(|metadata| metadata.len() == 0) { + return (Repaired::Empty, false); + } + let failed = |err: anyhow::Error| Repaired::Failed(format!("{err:#}")); + let finish = match read(path) { + Ok(Some(finish)) => finish, + Ok(None) => return (Repaired::Finished, false), + Err(err) => { + let again = io(&err); + return (failed(err), again); + } + }; + match copy(path, &finish) { + Ok(repaired) => (Repaired::Saved(repaired), false), + Err(err) => (failed(err), true), + } +} + +/// Whether `err` is a failure to read or write rather than what was read. +fn io(err: &anyhow::Error) -> bool { + err.chain().any(|err| { + err.is::() || matches!(err.downcast_ref::(), Some(moq_mux::Error::Io(_))) + }) +} + +/// The writes that finish `path`, or `None` when it is already a regular MP4. +fn read(path: &Path) -> anyhow::Result> { + let mut file = std::fs::File::open(path).with_context(|| format!("cannot open {}", path.display()))?; + mp4::repair(&mut file).context("not a recording the studio can repair") +} + +/// Copy `path` with `finish` applied to `.repaired.mp4` beside it, +/// numbered rather than replacing a file already there, such as the recording +/// of a source named like a repair. +fn copy(path: &Path, finish: &mp4::Finish) -> anyhow::Result { + let dir = path.parent().context("a recording has a directory")?; + let stem = stem(path)?; + // Written in a directory of its own first, which nothing else can name, so + // a repair cut short never leaves a file under a repaired name. Dropping + // it removes what is left. + let staging = tempfile::Builder::new() + .prefix(".repairing") + .tempdir_in(dir) + .with_context(|| format!("cannot create a directory in {}", dir.display()))?; + let partial = staging.path().join("repair.mp4"); + apply(path, &partial, finish).with_context(|| format!("cannot copy {}", path.display()))?; + let repaired = claim(dir, &stem)?; + // Replaces only the empty file just claimed. + std::fs::rename(&partial, &repaired) + .inspect_err(|_| remove(&repaired)) + .with_context(|| format!("cannot write {}", repaired.display()))?; + // The marker is removed next, so the new name reaches the disk first. + // Windows cannot open a directory to sync it. + #[cfg(unix)] + std::fs::File::open(dir) + .and_then(|dir| dir.sync_all()) + .with_context(|| format!("cannot sync {}", dir.display()))?; + Ok(repaired) +} + +/// The stem of the repair of the recording `path`: `.repaired`. +pub(super) fn stem(path: &Path) -> anyhow::Result { + let name = path.file_stem().and_then(|name| name.to_str()); + let name = name.with_context(|| format!("{} has no name", path.display()))?; + Ok(format!("{name}{}", super::REPAIRED)) +} + +/// Create an empty file named `stem` in `dir`, numbering it rather than +/// replacing one already there. +fn claim(dir: &Path, stem: &str) -> anyhow::Result { + let mut number: u32 = 1; + loop { + let path = dir.join(super::numbered(stem, number)); + match std::fs::File::create_new(&path) { + Ok(_) => return Ok(path), + 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())), + } + } +} + +/// Copy `path` to `to` and finish the copy with `finish`, as a recording +/// finishes: cut where the moov goes, the moov, then the mdat header. +fn apply(path: &Path, to: &Path, finish: &mp4::Finish) -> std::io::Result<()> { + // A clone on file systems that copy on write, so even a long recording + // repairs at once there. + std::fs::copy(path, to)?; + // The copy takes the original's permissions, and the original may be + // read-only. + let mut permissions = std::fs::metadata(to)?.permissions(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + permissions.set_mode(permissions.mode() | 0o200); + } + #[cfg(not(unix))] + #[allow( + clippy::permissions_set_readonly_false, + reason = "only clears the read-only attribute here" + )] + permissions.set_readonly(false); + std::fs::set_permissions(to, permissions)?; + let mut file = std::fs::OpenOptions::new().write(true).open(to)?; + file.set_len(finish.moov_offset)?; + file.seek(SeekFrom::Start(finish.moov_offset))?; + file.write_all(&finish.moov)?; + file.seek(SeekFrom::Start(finish.mdat_offset))?; + file.write_all(&finish.mdat)?; + file.sync_all() +} + +/// Remove `path`, which may already be gone. +fn remove(path: &Path) { + if let Err(err) = std::fs::remove_file(path) + && err.kind() != std::io::ErrorKind::NotFound + { + tracing::warn!(path = %path.display(), error = %err, "cannot remove"); + } +} diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index f548f3d..923a95b 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -2658,6 +2658,7 @@ impl Panels { /// How the stream reaches viewers: what is wrong and why, each relay, the /// uplink, each track, and each viewer that reports. fn health(&mut self, ui: &mut Ui) { + repairs(ui, &self.engine.recorder.repairs()); let report = self.engine.health.lock().unwrap(); if report.relays.is_empty() { ui.label("Not publishing; start with --connect to see how the stream is delivered."); @@ -2808,6 +2809,32 @@ impl Panels { } } +/// The recordings a crash left unfinished, found at launch, and how each +/// repair went. +fn repairs(ui: &mut Ui, repairs: &[record::Repair]) { + if repairs.is_empty() { + return; + } + let file = |path: &std::path::Path| path.file_name().unwrap_or_default().to_string_lossy().into_owned(); + for repair in repairs { + let name = file(&repair.path); + match &repair.outcome { + None => ui.weak(format!("Repairing {name}, which the studio left unfinished")), + Some(record::Repaired::Saved(repaired)) => ui.colored_label( + DEGRADED, + format!("{name} was left unfinished; repaired as {}", file(repaired)), + ), + Some(record::Repaired::Finished) => ui.weak(format!("{name} was already finished")), + Some(record::Repaired::Empty) => ui.weak(format!("{name} was left unfinished with nothing recorded")), + Some(record::Repaired::Failed(reason)) => { + ui.colored_label(ui.visuals().error_fg_color, format!("Cannot repair {name}: {reason}")) + } + } + .on_hover_text(repair.path.display().to_string()); + } + ui.separator(); +} + const HEALTHY: Color32 = Color32::from_rgb(80, 200, 120); const DEGRADED: Color32 = Color32::from_rgb(220, 180, 60);