From f0e023504d5587ec1f91a5fdfeea38fc6e39fcb5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 22:43:01 -0800 Subject: [PATCH 1/7] chore(quest): claim quest/m0/recording-repair.md Quest-Claim: ba73345cde8d4b80 From 557df3141034565aaf398e1df5c51a4c82a0b474 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 23:04:07 -0800 Subject: [PATCH 2/7] feat(studio): repair recordings a crash left unfinished at launch Each recorded file gets a locked sidecar marker, `.mp4.part`, removed once the file is a regular MP4 or removed. At launch the recorder repairs every file whose marker nobody holds into `.repaired.mp4` beside it, in the background, leaving the original untouched, and reports each repair in the Health panel and the log. Co-Authored-By: Claude Opus 5.5 --- README.md | 11 +- quest/m0/README.md | 1 - quest/m0/recording-repair.md | 33 ---- rs/moq-studio/src/record.rs | 288 ++++++++++++++++++++++++++--- rs/moq-studio/src/record/repair.rs | 212 +++++++++++++++++++++ rs/moq-studio/src/ui.rs | 27 +++ 6 files changed, 515 insertions(+), 57 deletions(-) delete mode 100644 quest/m0/recording-repair.md create mode 100644 rs/moq-studio/src/record/repair.rs diff --git a/README.md b/README.md index e5a3f4a..df0ce80 100644 --- a/README.md +++ b/README.md @@ -100,7 +100,7 @@ Every panel is a tab you can drag, split, or stack. | 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 | 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, and see the broadcast path, whether the program is encoding, the encoder sessions open, and each rendition's 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 @@ -632,6 +632,15 @@ 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.part` 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, 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 eb29ff6..cd7ca0e 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -105,7 +105,6 @@ repository. - [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 - [Clean quit](/quest/m0/clean-quit.md) - quitting finishes every track and closes the relay session, so the broadcast's tail arrives -- [Recover crashed recordings](/quest/m0/recording-repair.md) - a recording a crash left unfinished is repaired beside the original at launch - [Premultiplied source sampling](/quest/m0/premultiply.md) - scaled transparent sources draw without edge fringes - [Deterministic browser source startup](/quest/m0/cef-startup.md) - CEF on the CPU path paints its first frame without the GPU fallback hang - [A browser renderer that dies on its first page](/quest/m0/browser-first-crash.md) - a renderer lost during the first navigation fails its source 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/src/record.rs b/rs/moq-studio/src/record.rs index 48d0d07..80f3d42 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(); } } @@ -385,16 +421,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 +461,19 @@ 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, the one its repair is +/// written under. +const SUFFIX: usize = " (4294967295)".len() + REPAIRED.len() + PARTIAL.len(); + +/// What a file's marker adds to its name. +const MARKER: &str = ".part"; + +/// What a repaired file's name has in place of `.mp4`. +const REPAIRED: &str = ".repaired.mp4"; + +/// What a repair's name has added while it is written. +const PARTIAL: &str = ".tmp"; /// 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. @@ -464,8 +511,8 @@ fn truncate(text: &str, bytes: usize) -> &str { } /// 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. +fn create(dir: &Path, stem: &str) -> anyhow::Result<(PathBuf, std::fs::File, Marker)> { let mut number: u32 = 1; loop { let name = match number { @@ -474,7 +521,12 @@ fn create(dir: &Path, stem: &str) -> anyhow::Result<(PathBuf, std::fs::File)> { }; let path = dir.join(name); match std::fs::File::create_new(&path) { - Ok(file) => return Ok((path, file)), + Ok(file) => { + let marker = Marker::create(&path).inspect_err(|_| { + let _ = std::fs::remove_file(&path); + })?; + return Ok((path, file, marker)); + } Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => { number = number.checked_add(1).context("every numbered name is taken")?; } @@ -483,6 +535,49 @@ fn create(dir: &Path, stem: &str) -> anyhow::Result<(PathBuf, std::fs::File)> { } } +/// The sidecar `.mp4.part` 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` unfinished. + fn create(path: &Path) -> anyhow::Result { + let path = marker(path); + let file = std::fs::File::create(&path).with_context(|| format!("cannot create {}", path.display()))?; + // A file system without locks still marks the file; only another studio + // recording into the same directory as it launches goes unnoticed. + if let Err(err) = file.try_lock() { + tracing::debug!(path = %path.display(), error = %err, "cannot lock a recording's marker"); + } + Ok(Self { path, file }) + } + + /// Remove the marker. + fn clear(self) { + // Unlocked first, which Windows needs to remove it. + drop(self.file); + 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"); + } + } +} + +/// 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 +668,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 +686,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 +1436,149 @@ 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"); + 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(), 4, "{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.part"), 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(); + let launch = studio.recorder(crashed.path()); + kio::clock::timeout(Duration::from_secs(5), launch.repaired()) + .await + .expect("the repairs end"); + + let repairs = launch.repairs(); + let outcomes: Vec<_> = repairs + .iter() + .map(|repair| { + let name = repair.path.file_name().unwrap().to_string_lossy().into_owned(); + ( + name.rsplit_once('Z').map_or(name.clone(), |(_, end)| end.to_string()), + repair.outcome.clone(), + ) + }) + .collect(); + // In name order, where a space comes before a dot. + let [ + (finished, Some(Repaired::Finished)), + (mic, Some(Repaired::Saved(_))), + (program, Some(Repaired::Saved(_))), + ] = outcomes.as_slice() + else { + panic!("{outcomes:?}"); + }; + assert_eq!([finished.as_str(), mic, program], ["finished.mp4", " mic.mp4", ".mp4"]); + for repair in &repairs[1..] { + let Some(Repaired::Saved(repaired)) = &repair.outcome else { + unreachable!(); + }; + assert_eq!(repaired, &repair::repaired(&repair.path)); + 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(".part") { + 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() - 3 + 2); + + // 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.part"), b"").unwrap(); + std::fs::write(dir.path().join("junk.mp4"), b"not a recording").unwrap(); + std::fs::write(dir.path().join("junk.mp4.part"), b"").unwrap(); + std::fs::write(dir.path().join("gone.mp4.part"), 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.part"] + ); + } + /// A long scene file or source name is cut, between characters, so every /// file name fits, numbered ones included. #[test] @@ -1353,9 +1593,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, while written and once done. + let repaired = repair::repaired(&path); + for path in [marker(&path), repair::partial(&repaired), repaired, path] { + let name = path.file_name().unwrap().len(); + assert!(name <= MAX_NAME, "{name} bytes: {}", path.display()); + } } } } @@ -1393,9 +1637,9 @@ 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")); diff --git a/rs/moq-studio/src/record/repair.rs b/rs/moq-studio/src/record/repair.rs new file mode 100644 index 0000000..d684608 --- /dev/null +++ b/rs/moq-studio/src/record/repair.rs @@ -0,0 +1,212 @@ +//! 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, which is repaired into +//! a regular MP4 beside it, `.repaired.mp4`, leaving the original as it +//! was. 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; + +/// 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 writing the repair + /// failed. A failed 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 paths = unfinished(dir); + let list: Vec<_> = paths + .iter() + .map(|path| Repair { + path: path.clone(), + outcome: None, + }) + .collect(); + let list = Arc::new(Mutex::new(list)); + if !paths.is_empty() { + let list = list.clone(); + runtime.spawn_blocking(move || { + for (index, path) in paths.iter().enumerate() { + let outcome = repair(path); + 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 holds, in name +/// order. A marker whose file is gone is removed. +fn unfinished(dir: &Path) -> Vec { + 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 paths: Vec = entries + .filter_map(|entry| { + let marker = entry.ok()?.path(); + let name = marker.file_name()?.to_str()?.strip_suffix(super::MARKER)?; + if !name.ends_with(".mp4") || held(&marker) { + return None; + } + let recording = dir.join(name); + if !recording.exists() { + tracing::debug!(marker = %marker.display(), "removing the marker of a recording that is gone"); + remove(&marker); + return None; + } + Some(recording) + }) + .collect(); + paths.sort(); + paths +} + +/// Whether a running recording, such as another studio's into the same +/// directory, holds `marker`'s lock. +fn held(marker: &Path) -> bool { + let Ok(file) = std::fs::File::open(marker) else { + return false; + }; + matches!(file.try_lock_shared(), Err(std::fs::TryLockError::WouldBlock)) +} + +/// Repair the unfinished recording at `path`, reporting it in the log. +fn repair(path: &Path) -> Repaired { + let outcome = write(path); + 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, and remove its marker unless writing +/// failed, which the next launch tries again. +fn write(path: &Path) -> Repaired { + let outcome = if std::fs::metadata(path).is_ok_and(|metadata| metadata.len() == 0) { + Repaired::Empty + } else { + match read(path) { + Ok(Some(finish)) => { + let repaired = repaired(path); + if let Err(err) = copy(path, &repaired, &finish) { + return Repaired::Failed(format!("{err:#}")); + } + Repaired::Saved(repaired) + } + Ok(None) => Repaired::Finished, + Err(err) => Repaired::Failed(format!("{err:#}")), + } + }; + remove(&super::marker(path)); + outcome +} + +/// 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` to `repaired` with `finish` applied, through a temporary file +/// so a repair cut short never leaves a file under the repaired name. +fn copy(path: &Path, repaired: &Path, finish: &mp4::Finish) -> anyhow::Result<()> { + let partial = partial(repaired); + let written = apply(path, &partial, finish).and_then(|()| std::fs::rename(&partial, repaired)); + if written.is_err() { + remove(&partial); + } + written.with_context(|| format!("cannot write {}", repaired.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)?; + 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() +} + +/// Where the repair of the recording `path` goes: `.repaired.mp4`. +pub(super) fn repaired(path: &Path) -> PathBuf { + path.with_extension(super::REPAIRED.trim_start_matches('.')) +} + +/// Where the repair is written before it takes its name. +pub(super) fn partial(repaired: &Path) -> PathBuf { + let mut name = repaired.as_os_str().to_owned(); + name.push(super::PARTIAL); + name.into() +} + +/// 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 e11653f..dab954c 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -2280,6 +2280,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."); @@ -2430,6 +2431,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); From 4b18b0be0be00ceb9cabeb11c8345d6a002e944b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 23:10:56 -0800 Subject: [PATCH 3/7] fix(studio): a launch holds a marker's lock through its repair, and retries after I/O failures Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 39 ++++++++- rs/moq-studio/src/record/repair.rs | 135 ++++++++++++++++++----------- 2 files changed, 119 insertions(+), 55 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 80f3d42..cecf44c 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -558,16 +558,16 @@ impl Marker { Ok(Self { path, file }) } - /// Remove the marker. + /// 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) { - // Unlocked first, which Windows needs to remove it. - drop(self.file); 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); } } @@ -1579,6 +1579,39 @@ mod tests { ); } + /// 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.part"), 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.part"]); + 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] diff --git a/rs/moq-studio/src/record/repair.rs b/rs/moq-studio/src/record/repair.rs index d684608..e9750c5 100644 --- a/rs/moq-studio/src/record/repair.rs +++ b/rs/moq-studio/src/record/repair.rs @@ -2,10 +2,11 @@ //! //! 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, which is repaired into -//! a regular MP4 beside it, `.repaired.mp4`, leaving the original as it -//! was. Removing the marker after the repair keeps the file from being -//! repaired again. +//! 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`, 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}; @@ -14,6 +15,8 @@ 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 { @@ -32,8 +35,9 @@ pub enum Repaired { Finished, /// Nothing reached it before the studio stopped. Empty, - /// It holds no recording that can be repaired, or writing the repair - /// failed. A failed write leaves the marker, so the next launch tries again. + /// 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), } @@ -48,20 +52,20 @@ impl Repairs { /// /// Only the markers there now count, not those of recordings started later. pub fn start(dir: &Path, runtime: &tokio::runtime::Handle) -> Self { - let paths = unfinished(dir); - let list: Vec<_> = paths + let claimed = unfinished(dir); + let list: Vec<_> = claimed .iter() - .map(|path| Repair { + .map(|(path, _)| Repair { path: path.clone(), outcome: None, }) .collect(); let list = Arc::new(Mutex::new(list)); - if !paths.is_empty() { + if !claimed.is_empty() { let list = list.clone(); runtime.spawn_blocking(move || { - for (index, path) in paths.iter().enumerate() { - let outcome = repair(path); + for (index, (path, marker)) in claimed.into_iter().enumerate() { + let outcome = repair(&path, marker); list.lock().unwrap()[index].outcome = Some(outcome); } }); @@ -75,9 +79,9 @@ impl Repairs { } } -/// Each file in `dir` whose marker no running recording holds, in name -/// order. A marker whose file is gone is removed. -fn unfinished(dir: &Path) -> Vec { +/// 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. @@ -87,38 +91,57 @@ fn unfinished(dir: &Path) -> Vec { return Vec::new(); } }; - let mut paths: Vec = entries + let mut claimed: Vec<_> = entries .filter_map(|entry| { - let marker = entry.ok()?.path(); - let name = marker.file_name()?.to_str()?.strip_suffix(super::MARKER)?; - if !name.ends_with(".mp4") || held(&marker) { + 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)?; if !recording.exists() { - tracing::debug!(marker = %marker.display(), "removing the marker of a recording that is gone"); - remove(&marker); + tracing::debug!(marker = %marker.path.display(), "removing the marker of a recording that is gone"); + marker.clear(); return None; } - Some(recording) + Some((recording, marker)) }) .collect(); - paths.sort(); - paths + claimed.sort_by(|(a, _), (b, _)| a.cmp(b)); + claimed } -/// Whether a running recording, such as another studio's into the same -/// directory, holds `marker`'s lock. -fn held(marker: &Path) -> bool { - let Ok(file) = std::fs::File::open(marker) else { - return false; - }; - matches!(file.try_lock_shared(), Err(std::fs::TryLockError::WouldBlock)) +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 { + let file = std::fs::File::open(&path).ok()?; + 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"); + } + } + // A repair that ended as this opened it removed it before letting go. + if !path.exists() { + return None; + } + Some(Self { path, file }) + } } -/// Repair the unfinished recording at `path`, reporting it in the log. -fn repair(path: &Path) -> Repaired { - let outcome = write(path); +/// 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(), @@ -136,26 +159,34 @@ fn repair(path: &Path) -> Repaired { outcome } -/// Write the repair of `path` beside it, and remove its marker unless writing -/// failed, which the next launch tries again. -fn write(path: &Path) -> Repaired { - let outcome = if std::fs::metadata(path).is_ok_and(|metadata| metadata.len() == 0) { - Repaired::Empty - } else { - match read(path) { - Ok(Some(finish)) => { - let repaired = repaired(path); - if let Err(err) = copy(path, &repaired, &finish) { - return Repaired::Failed(format!("{err:#}")); - } - Repaired::Saved(repaired) - } - Ok(None) => Repaired::Finished, - Err(err) => Repaired::Failed(format!("{err:#}")), +/// 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); } }; - remove(&super::marker(path)); - outcome + let repaired = repaired(path); + match copy(path, &repaired, &finish) { + Ok(()) => (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. From 9ca1482212408cba4dfaf395f686263c6630cd79 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 06:52:59 -0800 Subject: [PATCH 4/7] fix(studio): a repair never replaces a file, and markers are the studio's own A repair is written in a directory of its own and published under a name it first claims with create_new, numbered when taken, so a source named like a repair (`mic` beside `mic.repaired`) keeps its recording. A recording's marker is created exclusively and locked before its file exists, so a launch that claims it first removes it rather than repairing a live recording. A launch opens a marker writable, as an exclusive lock needs on NFS. Markers end in `.unfinished` rather than `.part`, which downloads use. Co-Authored-By: Claude Opus 5.5 --- README.md | 11 +- rs/moq-studio/src/record.rs | 172 ++++++++++++++++++----------- rs/moq-studio/src/record/repair.rs | 82 +++++++++----- 3 files changed, 172 insertions(+), 93 deletions(-) diff --git a/README.md b/README.md index 6a1e228..d2d41af 100644 --- a/README.md +++ b/README.md @@ -741,11 +741,12 @@ 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.part` 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, 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 +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. diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index cecf44c..57c169e 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -462,18 +462,19 @@ impl Run { const MAX_NAME: usize = 255; /// The most a recording's files add to a stem: the largest number [`create`] -/// gives, and the longest name a file takes from there, the one its repair is -/// written under. -const SUFFIX: usize = " (4294967295)".len() + REPAIRED.len() + PARTIAL.len(); +/// 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(); -/// What a file's marker adds to its name. -const MARKER: &str = ".part"; +/// The largest number a taken name is given. +const NUMBER: &str = " (4294967295)"; -/// What a repaired file's name has in place of `.mp4`. -const REPAIRED: &str = ".repaired.mp4"; +/// 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 repair's name has added while it is written. -const PARTIAL: &str = ".tmp"; +/// 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. @@ -510,32 +511,35 @@ 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, and mark it unfinished. +/// 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) => { - let marker = Marker::create(&path).inspect_err(|_| { - let _ = std::fs::remove_file(&path); - })?; - return Ok((path, file, marker)); - } - 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())); + } } - Err(err) => return Err(err).with_context(|| format!("cannot create {}", path.display())), } + number = number.checked_add(1).context("every numbered name is taken")?; } } -/// The sidecar `.mp4.part` beside a file being recorded, until the file +/// 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. @@ -546,16 +550,34 @@ struct Marker { } impl Marker { - /// Mark the recording at `path` unfinished. - fn create(path: &Path) -> anyhow::Result { + /// 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 = std::fs::File::create(&path).with_context(|| format!("cannot create {}", path.display()))?; - // A file system without locks still marks the file; only another studio - // recording into the same directory as it launches goes unnoticed. - if let Err(err) = file.try_lock() { - tracing::debug!(path = %path.display(), error = %err, "cannot lock a recording's marker"); + 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. + if !path.exists() { + return Ok(None); } - Ok(Self { path, file }) + Ok(Some(Self { path, file })) } /// Remove the marker, and only then let go of its lock, so a launch that @@ -1456,13 +1478,15 @@ mod tests { 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(), 4, "{names_while_recording:?}"); + 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(); } @@ -1475,7 +1499,7 @@ mod tests { // 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.part"), b"").unwrap(); + std::fs::write(crashed.path().join("finished.mp4.unfinished"), b"").unwrap(); let originals: Vec<_> = names(crashed.path()) .into_iter() @@ -1489,32 +1513,37 @@ mod tests { .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 name = repair.path.file_name().unwrap().to_string_lossy().into_owned(); - ( - name.rsplit_once('Z').map_or(name.clone(), |(_, end)| end.to_string()), - repair.outcome.clone(), - ) + 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. - let [ - (finished, Some(Repaired::Finished)), - (mic, Some(Repaired::Saved(_))), - (program, Some(Repaired::Saved(_))), - ] = outcomes.as_slice() - else { - panic!("{outcomes:?}"); - }; - assert_eq!([finished.as_str(), mic, program], ["finished.mp4", " mic.mp4", ".mp4"]); + // 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!(); }; - assert_eq!(repaired, &repair::repaired(&repair.path)); let [(handler, duration)] = tracks(repaired).try_into().unwrap(); assert_eq!(handler, "soun"); assert!( @@ -1527,13 +1556,13 @@ mod tests { // 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(".part") { + 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() - 3 + 2); + assert_eq!(names(crashed.path()).len(), originals.len() - 4 + 3); // The next launch has nothing to repair. let again = studio.recorder(crashed.path()); @@ -1548,10 +1577,10 @@ mod tests { 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.part"), 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.part"), b"").unwrap(); - std::fs::write(dir.path().join("gone.mp4.part"), b"").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()); @@ -1575,7 +1604,7 @@ mod tests { assert!(reason.contains("not a recording the studio can repair"), "{reason}"); assert_eq!( names(dir.path()), - ["empty.mp4", "junk.mp4", "recording.mp4", "recording.mp4.part"] + ["empty.mp4", "junk.mp4", "recording.mp4", "recording.mp4.unfinished"] ); } @@ -1590,7 +1619,7 @@ mod tests { 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.part"), b"").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()); @@ -1606,7 +1635,7 @@ mod tests { let (reason, names) = launch(0o000).await; assert!(reason.contains("cannot open"), "{reason}"); - assert_eq!(names, ["locked.mp4", "locked.mp4.part"]); + 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"]); @@ -1627,9 +1656,9 @@ mod tests { assert!(source.len() >= program.len() + 1 + SOURCE - 1, "{source}"); for stem in [&program, &source, &source] { let (path, ..) = create(dir.path(), stem).unwrap(); - // So do its marker and its repair, while written and once done. - let repaired = repair::repaired(&path); - for path in [marker(&path), repair::partial(&repaired), repaired, path] { + // 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()); } @@ -1678,6 +1707,23 @@ mod tests { 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"] + ); + } + /// 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 index e9750c5..a0289cb 100644 --- a/rs/moq-studio/src/record/repair.rs +++ b/rs/moq-studio/src/record/repair.rs @@ -4,7 +4,8 @@ //! 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`, leaving the original as it was, so neither a +//! `.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. @@ -117,7 +118,16 @@ impl Marker { /// another studio's into the same directory, or another launch's repair /// holds it. fn claim(path: PathBuf) -> Option { - let file = std::fs::File::open(&path).ok()?; + // 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, @@ -175,9 +185,8 @@ fn write(path: &Path) -> (Repaired, bool) { return (failed(err), again); } }; - let repaired = repaired(path); - match copy(path, &repaired, &finish) { - Ok(()) => (Repaired::Saved(repaired), false), + match copy(path, &finish) { + Ok(repaired) => (Repaired::Saved(repaired), false), Err(err) => (failed(err), true), } } @@ -195,15 +204,50 @@ fn read(path: &Path) -> anyhow::Result> { mp4::repair(&mut file).context("not a recording the studio can repair") } -/// Copy `path` to `repaired` with `finish` applied, through a temporary file -/// so a repair cut short never leaves a file under the repaired name. -fn copy(path: &Path, repaired: &Path, finish: &mp4::Finish) -> anyhow::Result<()> { - let partial = partial(repaired); - let written = apply(path, &partial, finish).and_then(|()| std::fs::rename(&partial, repaired)); - if written.is_err() { - remove(&partial); +/// 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()))?; + 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())), + } } - written.with_context(|| format!("cannot write {}", repaired.display())) } /// Copy `path` to `to` and finish the copy with `finish`, as a recording @@ -221,18 +265,6 @@ fn apply(path: &Path, to: &Path, finish: &mp4::Finish) -> std::io::Result<()> { file.sync_all() } -/// Where the repair of the recording `path` goes: `.repaired.mp4`. -pub(super) fn repaired(path: &Path) -> PathBuf { - path.with_extension(super::REPAIRED.trim_start_matches('.')) -} - -/// Where the repair is written before it takes its name. -pub(super) fn partial(repaired: &Path) -> PathBuf { - let mut name = repaired.as_os_str().to_owned(); - name.push(super::PARTIAL); - name.into() -} - /// Remove `path`, which may already be gone. fn remove(path: &Path) { if let Err(err) = std::fs::remove_file(path) From c2330de5e2b18798eede616f4a7ce65487cb56dd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 07:04:09 -0800 Subject: [PATCH 5/7] fix(studio): a marker's holder checks its path still names it Each marker holds a token no other holds, and whoever locks one checks that its path still reads that token, so a recording or launch never takes, and then removes, a marker another recording created at the same name after the one it opened was removed. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 42 +++++++++++++++++++++++++++--- rs/moq-studio/src/record/repair.rs | 7 ++--- 2 files changed, 43 insertions(+), 6 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 57c169e..04ace1e 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -558,11 +558,20 @@ impl Marker { /// 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) { + let mut 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())), }; + let token = { + use std::hash::BuildHasher; + let random = std::hash::RandomState::new().hash_one(()); + format!("{} {random:016x}", std::process::id()) + }; + if let Err(err) = std::io::Write::write_all(&mut file, token.as_bytes()) { + let _ = std::fs::remove_file(&path); + return Err(err).with_context(|| format!("cannot write {}", path.display())); + } match file.try_lock() { Ok(()) => {} // A launch holds it, and removes it once it finds no recording. @@ -573,13 +582,27 @@ impl Marker { tracing::warn!(path = %path.display(), error = %err, "cannot lock a recording's marker"); } } - // A launch that held it before it was locked removed it. - if !path.exists() { + // 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, &mut 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. Each marker holds a token no other + /// holds, since file identity has no portable query. Whoever removes a + /// marker holds its lock first, so the answer stands while the lock is held. + fn named(path: &Path, file: &mut std::fs::File) -> bool { + use std::io::{Read, Seek}; + let mut held = Vec::new(); + if file.rewind().and_then(|()| file.read_to_end(&mut held)).is_err() { + return false; + } + std::fs::read(path).is_ok_and(|named| named == held) + } + /// 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) { @@ -1724,6 +1747,19 @@ mod tests { ); } + /// 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. + #[test] + fn a_replaced_marker_is_not_its_holders() { + let dir = tempfile::tempdir().unwrap(); + let recording = dir.path().join("show.mp4"); + let Marker { path, mut file } = Marker::create(&recording).unwrap().unwrap(); + assert!(Marker::named(&path, &mut file)); + std::fs::remove_file(&path).unwrap(); + let _other = Marker::create(&recording).unwrap().unwrap(); + assert!(!Marker::named(&path, &mut file)); + } + /// 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 index a0289cb..5e6182e 100644 --- a/rs/moq-studio/src/record/repair.rs +++ b/rs/moq-studio/src/record/repair.rs @@ -119,7 +119,7 @@ impl Marker { /// 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) { + let mut 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, @@ -137,8 +137,9 @@ impl Marker { tracing::debug!(path = %path.display(), error = %err, "cannot lock a recording's marker"); } } - // A repair that ended as this opened it removed it before letting go. - if !path.exists() { + // Whoever held it before removed it, and a new recording may have + // marked the same name since. + if !Self::named(&path, &mut file) { return None; } Some(Self { path, file }) From 9475db5d9d8a329019c67753b932f2b3a5805de2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 07:13:11 -0800 Subject: [PATCH 6/7] fix(studio): identify a locked marker by its file, not by reading it Windows refuses a read through another handle while the marker is locked, so every new marker failed the check. Compare file identities with same-file instead, which opens the path without data access, and fail loud when the identity cannot be read. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + rs/moq-studio/Cargo.toml | 3 +++ rs/moq-studio/src/record.rs | 39 ++++++++++++------------------ rs/moq-studio/src/record/repair.rs | 11 ++++++--- 4 files changed, 28 insertions(+), 26 deletions(-) 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/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 04ace1e..8874bde 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -558,20 +558,11 @@ impl Marker { /// finds no recording and removes it rather than repairing a live one. fn create(path: &Path) -> anyhow::Result> { let path = marker(path); - let mut file = match std::fs::File::create_new(&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())), }; - let token = { - use std::hash::BuildHasher; - let random = std::hash::RandomState::new().hash_one(()); - format!("{} {random:016x}", std::process::id()) - }; - if let Err(err) = std::io::Write::write_all(&mut file, token.as_bytes()) { - let _ = std::fs::remove_file(&path); - return Err(err).with_context(|| format!("cannot write {}", path.display())); - } match file.try_lock() { Ok(()) => {} // A launch holds it, and removes it once it finds no recording. @@ -584,23 +575,25 @@ impl 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, &mut file) { + 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. Each marker holds a token no other - /// holds, since file identity has no portable query. Whoever removes a - /// marker holds its lock first, so the answer stands while the lock is held. - fn named(path: &Path, file: &mut std::fs::File) -> bool { - use std::io::{Read, Seek}; - let mut held = Vec::new(); - if file.rewind().and_then(|()| file.read_to_end(&mut held)).is_err() { - return false; + /// 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())), } - std::fs::read(path).is_ok_and(|named| named == held) } /// Remove the marker, and only then let go of its lock, so a launch that @@ -1753,11 +1746,11 @@ mod tests { fn a_replaced_marker_is_not_its_holders() { let dir = tempfile::tempdir().unwrap(); let recording = dir.path().join("show.mp4"); - let Marker { path, mut file } = Marker::create(&recording).unwrap().unwrap(); - assert!(Marker::named(&path, &mut file)); + 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, &mut file)); + assert!(!Marker::named(&path, &file).unwrap()); } /// Starting into a directory that does not exist fails at once. diff --git a/rs/moq-studio/src/record/repair.rs b/rs/moq-studio/src/record/repair.rs index 5e6182e..89da31e 100644 --- a/rs/moq-studio/src/record/repair.rs +++ b/rs/moq-studio/src/record/repair.rs @@ -119,7 +119,7 @@ impl Marker { /// holds it. fn claim(path: PathBuf) -> Option { // Writable, as an exclusive lock needs on NFS. - let mut file = match std::fs::OpenOptions::new().read(true).write(true).open(&path) { + 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, @@ -139,8 +139,13 @@ impl Marker { } // Whoever held it before removed it, and a new recording may have // marked the same name since. - if !Self::named(&path, &mut file) { - return None; + 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 }) } From f0683b2d59dd9c600c681e8d9ebf3448538842fb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 08:30:57 -0800 Subject: [PATCH 7/7] fix(studio): repairs survive a read-only original and a power loss A repair makes its staging copy writable, since the copy takes a read-only original's permissions, and syncs the directory after publishing it, before the marker goes. A launch keeps a marker it cannot tell is orphaned. A recording that fails lets go of its marker's lock, so another studio's launch can repair it. The marker identity test runs on Unix only, as Windows keeps a removed name taken while a handle to it is open. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/record.rs | 13 ++++++++++ rs/moq-studio/src/record/repair.rs | 38 ++++++++++++++++++++++++++---- 2 files changed, 46 insertions(+), 5 deletions(-) diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index 8874bde..37b0b05 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -281,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()); } } @@ -1524,6 +1528,13 @@ mod tests { (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 @@ -1742,6 +1753,8 @@ mod tests { /// 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(); diff --git a/rs/moq-studio/src/record/repair.rs b/rs/moq-studio/src/record/repair.rs index 89da31e..46d45f4 100644 --- a/rs/moq-studio/src/record/repair.rs +++ b/rs/moq-studio/src/record/repair.rs @@ -101,12 +101,19 @@ fn unfinished(dir: &Path) -> Vec<(PathBuf, Marker)> { } let recording = dir.join(name); let marker = Marker::claim(path)?; - if !recording.exists() { - tracing::debug!(marker = %marker.path.display(), "removing the marker of a recording that is gone"); - marker.clear(); - return None; + 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 + } } - Some((recording, marker)) }) .collect(); claimed.sort_by(|(a, _), (b, _)| a.cmp(b)); @@ -230,6 +237,12 @@ fn copy(path: &Path, finish: &mp4::Finish) -> anyhow::Result { 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) } @@ -262,6 +275,21 @@ 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))?;