diff --git a/crates/enc-ffmpeg/src/mux/fragment_metadata.rs b/crates/enc-ffmpeg/src/mux/fragment_metadata.rs index d9b3159a265..64dcbed3d89 100644 --- a/crates/enc-ffmpeg/src/mux/fragment_metadata.rs +++ b/crates/enc-ffmpeg/src/mux/fragment_metadata.rs @@ -13,7 +13,23 @@ pub(super) struct FragmentMetadata { pub(super) fn read_fragment_metadata(path: &Path) -> io::Result { let mut file = File::open(path)?; let file_size = file.metadata()?.len(); - read_metadata(&mut file, file_size) + match read_metadata(&mut file, file_size) { + Ok(metadata) => Ok(metadata), + Err(error) => { + let init = path.with_file_name("init.mp4"); + if !init.is_file() { + return Err(error); + } + let prefix = super::fragment_prefix::read_fragment_prefix(&init, path)?; + if prefix.length != file_size { + return Err(invalid_fragment("Fragment contains incomplete media")); + } + Ok(FragmentMetadata { + duration: prefix.duration, + file_size, + }) + } + } } fn invalid_fragment(message: &'static str) -> io::Error { diff --git a/crates/enc-ffmpeg/src/mux/fragment_prefix.rs b/crates/enc-ffmpeg/src/mux/fragment_prefix.rs new file mode 100644 index 00000000000..2615ca88887 --- /dev/null +++ b/crates/enc-ffmpeg/src/mux/fragment_prefix.rs @@ -0,0 +1,615 @@ +use std::{ + fs::File, + io::{self, BufReader, Read, Seek, SeekFrom}, + path::Path, + time::Duration, +}; + +struct Reader { + input: BufReader, + position: u64, +} + +impl Reader { + fn open(path: &Path) -> io::Result { + Ok(Self { + input: BufReader::new(File::open(path)?), + position: 0, + }) + } + + fn len(&self) -> io::Result { + Ok(self.input.get_ref().metadata()?.len()) + } +} + +impl Read for Reader { + fn read(&mut self, bytes: &mut [u8]) -> io::Result { + let read = self.input.read(bytes)?; + self.position += read as u64; + Ok(read) + } +} + +impl Seek for Reader { + fn seek(&mut self, position: SeekFrom) -> io::Result { + if let SeekFrom::Current(0) = position { + return Ok(self.position); + } + self.position = self.input.seek(position)?; + Ok(self.position) + } +} + +#[derive(Clone, Copy)] +struct BoxHeader { + kind: [u8; 4], + start: u64, + payload: u64, + end: u64, +} + +#[derive(Default)] +struct Track { + id: u32, + timescale: u32, + duration: u32, + size: u32, + flags: u32, +} + +struct Run { + start: u64, + end: u64, +} + +struct Fragment { + decode_time: u64, + duration: u64, + runs: Vec, + starts_with_keyframe: bool, +} + +pub struct FragmentPrefix { + pub length: u64, + pub duration: Duration, +} + +fn invalid() -> io::Error { + io::Error::new( + io::ErrorKind::InvalidData, + "Invalid fragmented MP4 sample table", + ) +} + +fn u32_at(reader: &mut (impl Read + Seek), end: u64) -> io::Result { + if reader + .stream_position()? + .checked_add(4) + .ok_or_else(invalid)? + > end + { + return Err(invalid()); + } + let mut bytes = [0; 4]; + reader.read_exact(&mut bytes)?; + Ok(u32::from_be_bytes(bytes)) +} + +fn u64_at(reader: &mut (impl Read + Seek), end: u64) -> io::Result { + Ok((u64::from(u32_at(reader, end)?) << 32) | u64::from(u32_at(reader, end)?)) +} + +fn skip(reader: &mut (impl Read + Seek), bytes: u64, end: u64) -> io::Result<()> { + let position = reader + .stream_position()? + .checked_add(bytes) + .ok_or_else(invalid)?; + if position > end { + return Err(invalid()); + } + reader.seek(SeekFrom::Start(position))?; + Ok(()) +} + +fn header(reader: &mut (impl Read + Seek), end: u64) -> io::Result> { + let start = reader.stream_position()?; + if start == end { + return Ok(None); + } + let size = u32_at(reader, end)?; + let kind = u32_at(reader, end)?.to_be_bytes(); + let size = match size { + 0 => return Err(invalid()), + 1 => u64_at(reader, end)?, + size => u64::from(size), + }; + let payload = reader.stream_position()?; + let box_end = start.checked_add(size).ok_or_else(invalid)?; + if box_end < payload || box_end > end { + return Err(invalid()); + } + Ok(Some(BoxHeader { + kind, + start, + payload, + end: box_end, + })) +} + +fn track_in(reader: &mut (impl Read + Seek), end: u64, track: &mut Track) -> io::Result<()> { + while let Some(item) = header(reader, end)? { + match &item.kind { + b"trak" | b"mdia" => track_in(reader, item.end, track)?, + b"tkhd" => { + let version = u32_at(reader, item.end)? >> 24; + skip( + reader, + match version { + 0 => 8, + 1 => 16, + _ => return Err(invalid()), + }, + item.end, + )?; + track.id = u32_at(reader, item.end)?; + } + b"mdhd" => { + let version = u32_at(reader, item.end)? >> 24; + skip( + reader, + match version { + 0 => 8, + 1 => 16, + _ => return Err(invalid()), + }, + item.end, + )?; + track.timescale = u32_at(reader, item.end)?; + } + _ => {} + } + reader.seek(SeekFrom::Start(item.end))?; + } + Ok(()) +} + +fn read_track(path: &Path) -> io::Result { + let mut reader = Reader::open(path)?; + let end = reader.len()?; + let mut track = Track::default(); + let mut defaults = Vec::new(); + while let Some(item) = header(&mut reader, end)? { + if item.kind == *b"moov" { + while let Some(child) = header(&mut reader, item.end)? { + match &child.kind { + b"trak" => { + if track.id != 0 { + return Err(invalid()); + } + track_in(&mut reader, child.end, &mut track)?; + } + b"mvex" => { + while let Some(entry) = header(&mut reader, child.end)? { + if entry.kind == *b"trex" { + if u32_at(&mut reader, entry.end)? != 0 { + return Err(invalid()); + } + let id = u32_at(&mut reader, entry.end)?; + let _description = u32_at(&mut reader, entry.end)?; + defaults.push(( + id, + u32_at(&mut reader, entry.end)?, + u32_at(&mut reader, entry.end)?, + u32_at(&mut reader, entry.end)?, + )); + } + reader.seek(SeekFrom::Start(entry.end))?; + } + } + _ => {} + } + reader.seek(SeekFrom::Start(child.end))?; + } + } + reader.seek(SeekFrom::Start(item.end))?; + } + if track.id == 0 || track.timescale == 0 { + return Err(invalid()); + } + let mut matching = defaults.into_iter().filter(|(id, ..)| *id == track.id); + if let Some((_, duration, size, flags)) = matching.next() { + track.duration = duration; + track.size = size; + track.flags = flags; + } + if matching.next().is_some() { + return Err(invalid()); + } + Ok(track) +} + +fn read_fragment( + reader: &mut (impl Read + Seek), + moof: BoxHeader, + track: &Track, +) -> io::Result { + let mut traf = None; + while let Some(item) = header(reader, moof.end)? { + if item.kind == *b"traf" && traf.replace(item).is_some() { + return Err(invalid()); + } + reader.seek(SeekFrom::Start(item.end))?; + } + let traf = traf.ok_or_else(invalid)?; + reader.seek(SeekFrom::Start(traf.payload))?; + let mut base = moof.start; + let mut default_duration = track.duration; + let mut default_size = track.size; + let mut default_flags = track.flags; + let mut decode_time = None; + let mut has_header = false; + let mut runs = Vec::new(); + while let Some(item) = header(reader, traf.end)? { + match &item.kind { + b"tfhd" => { + if has_header { + return Err(invalid()); + } + has_header = true; + let flags = u32_at(reader, item.end)?; + if flags & !0x02003b != 0 || u32_at(reader, item.end)? != track.id { + return Err(invalid()); + } + if flags & 1 != 0 { + base = u64_at(reader, item.end)?; + } + if flags & 2 != 0 { + let _description = u32_at(reader, item.end)?; + } + if flags & 8 != 0 { + default_duration = u32_at(reader, item.end)?; + } + if flags & 16 != 0 { + default_size = u32_at(reader, item.end)?; + } + if flags & 32 != 0 { + default_flags = u32_at(reader, item.end)?; + } + } + b"tfdt" => { + if decode_time.is_some() { + return Err(invalid()); + } + decode_time = Some(match u32_at(reader, item.end)? { + 0 => u64::from(u32_at(reader, item.end)?), + 0x01000000 => u64_at(reader, item.end)?, + _ => return Err(invalid()), + }); + } + b"trun" => runs.push(item), + _ => {} + } + reader.seek(SeekFrom::Start(item.end))?; + } + if !has_header || runs.is_empty() { + return Err(invalid()); + } + let mut fragment = Fragment { + decode_time: decode_time.ok_or_else(invalid)?, + duration: 0, + runs: Vec::new(), + starts_with_keyframe: false, + }; + let mut next_data = base; + for run in runs { + reader.seek(SeekFrom::Start(run.payload))?; + let flags = u32_at(reader, run.end)?; + if flags & !0x01000f05 != 0 || flags >> 24 > 1 { + return Err(invalid()); + } + let count = u32_at(reader, run.end)?; + if count == 0 { + return Err(invalid()); + } + if flags & 1 != 0 { + let offset = u32_at(reader, run.end)? as i32; + next_data = base + .checked_add_signed(i64::from(offset)) + .ok_or_else(invalid)?; + } + let first_flags = if flags & 4 != 0 { + u32_at(reader, run.end)? + } else { + default_flags + }; + let per_sample = (flags & 0xf00).count_ones() as u64 * 4; + if per_sample + .checked_mul(u64::from(count)) + .ok_or_else(invalid)? + > run.end - reader.stream_position()? + { + return Err(invalid()); + } + let start = next_data; + if per_sample == 0 { + if default_duration == 0 || default_size == 0 { + return Err(invalid()); + } + if fragment.runs.is_empty() { + fragment.starts_with_keyframe = first_flags & 0x10000 == 0; + } + fragment.duration = fragment + .duration + .checked_add(u64::from(default_duration) * u64::from(count)) + .ok_or_else(invalid)?; + next_data = next_data + .checked_add(u64::from(default_size) * u64::from(count)) + .ok_or_else(invalid)?; + } + for index in 0..if per_sample == 0 { 0 } else { count } { + let duration = if flags & 0x100 != 0 { + u32_at(reader, run.end)? + } else { + default_duration + }; + let size = if flags & 0x200 != 0 { + u32_at(reader, run.end)? + } else { + default_size + }; + let sample_flags = if flags & 0x400 != 0 { + u32_at(reader, run.end)? + } else if index == 0 { + first_flags + } else { + default_flags + }; + if flags & 0x800 != 0 { + let _composition_offset = u32_at(reader, run.end)?; + } + if duration == 0 || size == 0 { + return Err(invalid()); + } + if fragment.runs.is_empty() && index == 0 { + fragment.starts_with_keyframe = sample_flags & 0x10000 == 0; + } + fragment.duration = fragment + .duration + .checked_add(u64::from(duration)) + .ok_or_else(invalid)?; + next_data = next_data.checked_add(u64::from(size)).ok_or_else(invalid)?; + } + if reader.stream_position()? != run.end { + return Err(invalid()); + } + fragment.runs.push(Run { + start, + end: next_data, + }); + } + Ok(fragment) +} + +fn is_incomplete_tail(error: &io::Error) -> bool { + matches!( + error.kind(), + io::ErrorKind::InvalidData | io::ErrorKind::UnexpectedEof + ) +} + +pub fn read_fragment_prefix(init: &Path, media: &Path) -> io::Result { + let track = read_track(init)?; + let mut reader = Reader::open(media)?; + let end = reader.len()?; + let mut pending = None; + let mut prefix = 0; + let mut first_time = None; + let mut end_time = 0; + let result = (|| -> io::Result<()> { + while let Some(item) = header(&mut reader, end)? { + match &item.kind { + b"moof" => { + if pending.is_some() { + return Err(invalid()); + } + pending = Some(read_fragment(&mut reader, item, &track)?); + } + b"mdat" => { + let fragment: Fragment = pending.take().ok_or_else(invalid)?; + if fragment + .runs + .iter() + .any(|run| run.start < item.payload || run.end > item.end) + || fragment + .runs + .windows(2) + .any(|pair| pair[0].end != pair[1].start) + || fragment + .runs + .first() + .is_none_or(|run| run.start != item.payload) + || fragment.runs.last().is_none_or(|run| run.end != item.end) + { + return Err(invalid()); + } + if first_time.is_none() { + if !fragment.starts_with_keyframe { + return Err(invalid()); + } + first_time = Some(fragment.decode_time); + } else if fragment.decode_time != end_time { + return Err(invalid()); + } + end_time = fragment + .decode_time + .checked_add(fragment.duration) + .ok_or_else(invalid)?; + prefix = item.end; + } + b"styp" | b"sidx" | b"emsg" | b"prft" | b"free" => { + if pending.is_some() { + return Err(invalid()); + } + } + _ => return Err(invalid()), + } + reader.seek(SeekFrom::Start(item.end))?; + } + Ok(()) + })(); + // Only a malformed or truncated tail ends the prefix; any other read + // error leaves the recording retryable instead of shortening it. + if let Err(error) = result + && !is_incomplete_tail(&error) + { + return Err(error); + } + if prefix == 0 { + return Err(invalid()); + } + let ticks = end_time + .checked_sub(first_time.ok_or_else(invalid)?) + .ok_or_else(invalid)?; + let timescale = u64::from(track.timescale); + Ok(FragmentPrefix { + length: prefix, + duration: Duration::new( + ticks / timescale, + ((ticks % timescale) * 1_000_000_000 / timescale) as u32, + ), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_malformed_or_truncated_tails_end_a_prefix() { + assert!(is_incomplete_tail(&invalid())); + assert!(is_incomplete_tail(&io::Error::from( + io::ErrorKind::UnexpectedEof + ))); + for kind in [ + io::ErrorKind::PermissionDenied, + io::ErrorKind::Interrupted, + io::ErrorKind::Other, + ] { + assert!(!is_incomplete_tail(&io::Error::from(kind))); + } + } + + fn boxed(kind: &[u8; 4], data: &[u8]) -> Vec { + let mut bytes = ((data.len() + 8) as u32).to_be_bytes().to_vec(); + bytes.extend(kind); + bytes.extend(data); + bytes + } + + fn words(values: &[u32]) -> Vec { + values + .iter() + .flat_map(|value| value.to_be_bytes()) + .collect() + } + + fn init() -> Vec { + let tkhd = boxed(b"tkhd", &words(&[0, 0, 0, 1])); + let mdhd = boxed(b"mdhd", &words(&[0, 0, 0, 30])); + let trak = boxed(b"trak", &[tkhd, boxed(b"mdia", &mdhd)].concat()); + let defaults = boxed(b"mvex", &boxed(b"trex", &words(&[0, 1, 1, 1, 4, 0]))); + boxed(b"moov", &[trak, defaults].concat()) + } + + fn media(time: u32, keyframe: bool, count: u32) -> Vec { + let tfhd = boxed(b"tfhd", &words(&[0x20000, 1])); + let tfdt = boxed(b"tfdt", &words(&[0, time])); + let make_moof = |offset| { + let trun = boxed( + b"trun", + &words(&[5, count, offset, if keyframe { 0 } else { 0x10000 }]), + ); + boxed( + b"moof", + &boxed(b"traf", &[tfhd.clone(), tfdt.clone(), trun].concat()), + ) + }; + let moof = make_moof(make_moof(0).len() as u32 + 8); + [moof, boxed(b"mdat", &vec![42; count as usize * 4])].concat() + } + + fn parse(init: &[u8], media: &[u8]) -> io::Result { + let directory = tempfile::tempdir()?; + let init_path = directory.path().join("init.mp4"); + let media_path = directory.path().join("segment.m4s"); + std::fs::write(&init_path, init)?; + std::fs::write(&media_path, media)?; + read_fragment_prefix(&init_path, &media_path) + } + + #[test] + fn accepts_multiple_fragments_without_indexes_and_with_multiple_indexes() { + for indexed in [false, true] { + let mut bytes = boxed(b"styp", &[0; 16]); + for index in 0..3 { + if indexed { + bytes.extend(boxed(b"sidx", &[0; 24])); + } + bytes.extend(media(index * 60, index == 0, 60)); + } + let prefix = parse(&init(), &bytes).unwrap(); + assert_eq!(prefix.length, bytes.len() as u64); + assert_eq!(prefix.duration, Duration::from_secs(6)); + } + } + + #[test] + fn every_truncated_tail_keeps_only_preceding_complete_samples() { + let first = media(0, true, 60); + let second = media(60, false, 60); + for cut in 0..second.len() { + let bytes = [&first[..], &second[..cut]].concat(); + let prefix = parse(&init(), &bytes).unwrap(); + assert_eq!(prefix.length, first.len() as u64, "cut {cut}"); + assert_eq!(prefix.duration, Duration::from_secs(2)); + } + for cut in 0..first.len() { + assert!(parse(&init(), &first[..cut]).is_err(), "cut {cut}"); + } + } + + #[test] + fn requires_keyframe_at_segment_start_and_contiguous_decode_times() { + assert!(parse(&init(), &media(0, false, 60)).is_err()); + for time in [0, 59, 61] { + let first = media(0, true, 60); + let bytes = [first.clone(), media(time, false, 60)].concat(); + assert_eq!(parse(&init(), &bytes).unwrap().length, first.len() as u64); + } + } + + #[test] + fn rejects_invalid_track_and_sample_ranges() { + let original = media(0, true, 60); + for (needle, delta, value) in [ + (b"tfhd", 8, 2), + (b"trun", 8, 0), + (b"trun", 12, 0), + (b"trun", 8, u32::MAX), + ] { + let mut bytes = original.clone(); + let offset = bytes.windows(4).position(|slice| slice == needle).unwrap() + delta; + bytes[offset..offset + 4].copy_from_slice(&value.to_be_bytes()); + assert!(parse(&init(), &bytes).is_err()); + } + let mut invalid_init = init(); + let offset = invalid_init + .windows(4) + .position(|slice| slice == b"mdhd") + .unwrap() + + 16; + invalid_init[offset..offset + 4].fill(0); + assert!(parse(&invalid_init, &original).is_err()); + } +} diff --git a/crates/enc-ffmpeg/src/mux/fragmented_mp4.rs b/crates/enc-ffmpeg/src/mux/fragmented_mp4.rs index 3361d0777c0..a126237365e 100644 --- a/crates/enc-ffmpeg/src/mux/fragmented_mp4.rs +++ b/crates/enc-ffmpeg/src/mux/fragmented_mp4.rs @@ -1,3 +1,4 @@ +pub use super::fragment_prefix::{FragmentPrefix, read_fragment_prefix}; use std::{ fs::File, io::{Read, Seek, SeekFrom}, diff --git a/crates/enc-ffmpeg/src/mux/mod.rs b/crates/enc-ffmpeg/src/mux/mod.rs index 24e53ac50fd..c31ea535e38 100644 --- a/crates/enc-ffmpeg/src/mux/mod.rs +++ b/crates/enc-ffmpeg/src/mux/mod.rs @@ -2,6 +2,7 @@ pub mod dash_audio; mod dash_output; pub mod fragment_manifest; mod fragment_metadata; +mod fragment_prefix; pub mod fragmented_audio; pub mod fragmented_mp4; pub mod mov; diff --git a/crates/enc-ffmpeg/src/mux/segmented_stream.rs b/crates/enc-ffmpeg/src/mux/segmented_stream.rs index 3b6091ceb5f..d252a921c6c 100644 --- a/crates/enc-ffmpeg/src/mux/segmented_stream.rs +++ b/crates/enc-ffmpeg/src/mux/segmented_stream.rs @@ -197,6 +197,15 @@ impl SegmentedVideoEncoder { base_path: PathBuf, video_config: VideoInfo, config: SegmentedVideoEncoderConfig, + ) -> Result { + Self::init_with_timed_fragments(base_path, video_config, config, false) + } + + pub fn init_with_timed_fragments( + base_path: PathBuf, + video_config: VideoInfo, + config: SegmentedVideoEncoderConfig, + timed_fragments: bool, ) -> Result { std::fs::create_dir_all(&base_path)?; @@ -231,6 +240,11 @@ impl SegmentedVideoEncoder { set_opt("use_template", "1"); set_opt("single_file", "0"); set_opt("hls_playlist", "1"); + if timed_fragments { + set_opt("frag_type", "duration"); + set_opt("frag_duration", "2"); + set_opt("streaming", "1"); + } } let mut builder = H264EncoderBuilder::new(video_config) diff --git a/crates/recording/Cargo.toml b/crates/recording/Cargo.toml index 20f858ac081..6ddb0dd6088 100644 --- a/crates/recording/Cargo.toml +++ b/crates/recording/Cargo.toml @@ -90,6 +90,7 @@ windows = { workspace = true, features = [ "Win32_UI_WindowsAndMessaging", "Win32_System_Performance", "Win32_Storage_Xps", + "Win32_Storage_FileSystem", ] } cap-d3d-adapter = { path = "../d3d-adapter" } diff --git a/crates/recording/src/capture_pipeline.rs b/crates/recording/src/capture_pipeline.rs index c55cf72e2c2..d935e0e30f2 100644 --- a/crates/recording/src/capture_pipeline.rs +++ b/crates/recording/src/capture_pipeline.rs @@ -177,6 +177,7 @@ impl MakeCapturePipeline for screen_capture::CMSampleBufferCapture { .with_timestamps(start_time) .with_start_gate(start_gate.clone()) .build::(MacOSFragmentedM4SMuxerConfig { + timed_fragments: true, preset, bpp, output_size, @@ -218,6 +219,7 @@ impl MakeCapturePipeline for screen_capture::CMSampleBufferCapture { .with_timestamps(start_time) .with_start_gate(start_gate.clone()) .build::(MacOSFragmentedM4SMuxerConfig { + timed_fragments: false, bpp: H264EncoderBuilder::INSTANT_MODE_BPP, output_size: Some(output_size), segment_tx, @@ -310,6 +312,7 @@ impl MakeCapturePipeline for screen_capture::Direct3DCapture { .with_timestamps(start_time) .with_start_gate(start_gate.clone()) .build::(WindowsFragmentedM4SMuxerConfig { + timed_fragments: true, segment_duration: std::time::Duration::from_secs(2), preset, bpp, @@ -358,6 +361,7 @@ impl MakeCapturePipeline for screen_capture::Direct3DCapture { .with_timestamps(start_time) .with_start_gate(start_gate.clone()) .build::(WindowsFragmentedM4SMuxerConfig { + timed_fragments: false, segment_duration: std::time::Duration::from_secs(2), preset: H264Preset::Ultrafast, bpp: H264EncoderBuilder::INSTANT_MODE_BPP, @@ -394,6 +398,7 @@ impl MakeCapturePipeline for screen_capture::X11Capture { .with_timestamps(start_time) .with_start_gate(start_gate.clone()) .build::(crate::ffmpeg::SegmentedVideoMuxerConfig { + timed_fragments: true, segment_duration: std::time::Duration::from_secs(2), preset: if ultra { H264Preset::Medium @@ -420,6 +425,7 @@ impl MakeCapturePipeline for screen_capture::X11Capture { .with_timestamps(start_time) .with_start_gate(start_gate.clone()) .build::(crate::ffmpeg::SegmentedVideoMuxerConfig { + timed_fragments: false, segment_duration: std::time::Duration::from_secs(2), preset: H264Preset::Ultrafast, output_size: Some(output_size), diff --git a/crates/recording/src/instant_recording.rs b/crates/recording/src/instant_recording.rs index 759025f1f01..18776eaaa64 100644 --- a/crates/recording/src/instant_recording.rs +++ b/crates/recording/src/instant_recording.rs @@ -764,6 +764,7 @@ async fn create_pipeline( .with_timestamps(start_time) .with_start_gate(start_gate.clone()) .build::(crate::ffmpeg::SegmentedVideoMuxerConfig { + timed_fragments: false, segment_duration: std::time::Duration::from_secs(2), preset: cap_enc_ffmpeg::h264::H264Preset::Ultrafast, output_size: Some(output_resolution), diff --git a/crates/recording/src/output_pipeline/ffmpeg.rs b/crates/recording/src/output_pipeline/ffmpeg.rs index 1c2f6cbd0e2..013528c27bb 100644 --- a/crates/recording/src/output_pipeline/ffmpeg.rs +++ b/crates/recording/src/output_pipeline/ffmpeg.rs @@ -444,6 +444,7 @@ struct SegmentedEncoderState { } pub struct SegmentedVideoMuxer { + timed_fragments: bool, base_path: PathBuf, video_config: VideoInfo, segment_duration: Duration, @@ -458,6 +459,7 @@ pub struct SegmentedVideoMuxer { } pub struct SegmentedVideoMuxerConfig { + pub timed_fragments: bool, pub segment_duration: Duration, pub preset: H264Preset, pub output_size: Option<(u32, u32)>, @@ -468,6 +470,7 @@ pub struct SegmentedVideoMuxerConfig { impl Default for SegmentedVideoMuxerConfig { fn default() -> Self { Self { + timed_fragments: false, segment_duration: Duration::from_secs(3), preset: H264Preset::Ultrafast, output_size: None, @@ -502,6 +505,7 @@ impl Muxer for SegmentedVideoMuxer { .unwrap_or_else(|| SharedPauseState::new(pause_flag)); Ok(Self { + timed_fragments: config.timed_fragments, base_path: output_path, video_config, segment_duration: config.segment_duration, @@ -665,8 +669,12 @@ impl SegmentedVideoMuxer { }; let slow_threshold_ms = frame_timing_log_threshold_ms(&self.video_config); - let mut encoder = - SegmentedVideoEncoder::init(self.base_path.clone(), self.video_config, encoder_config)?; + let mut encoder = SegmentedVideoEncoder::init_with_timed_fragments( + self.base_path.clone(), + self.video_config, + encoder_config, + self.timed_fragments, + )?; if let Some(tx) = &self.segment_tx { encoder.set_segment_callback(tx.clone()); } diff --git a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs index 6c490b64e6f..d2c6f7cec34 100644 --- a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs @@ -154,6 +154,7 @@ struct EncoderState { } pub struct MacOSFragmentedM4SMuxer { + timed_fragments: bool, base_path: PathBuf, video_config: VideoInfo, segment_duration: Duration, @@ -170,6 +171,7 @@ pub struct MacOSFragmentedM4SMuxer { } pub struct MacOSFragmentedM4SMuxerConfig { + pub timed_fragments: bool, pub segment_duration: Duration, pub preset: H264Preset, pub bpp: f32, @@ -182,6 +184,7 @@ pub struct MacOSFragmentedM4SMuxerConfig { impl Default for MacOSFragmentedM4SMuxerConfig { fn default() -> Self { Self { + timed_fragments: false, segment_duration: Duration::from_secs(2), preset: H264Preset::Ultrafast, bpp: H264EncoderBuilder::QUALITY_BPP, @@ -218,6 +221,7 @@ impl Muxer for MacOSFragmentedM4SMuxer { .unwrap_or_else(|| SharedPauseState::new(pause_flag)); Ok(Self { + timed_fragments: config.timed_fragments, base_path: output_path, video_config, segment_duration: config.segment_duration, @@ -295,8 +299,12 @@ impl MacOSFragmentedM4SMuxer { prefer_videotoolbox_hw_input: true, }; - let mut encoder = - SegmentedVideoEncoder::init(self.base_path.clone(), self.video_config, encoder_config)?; + let mut encoder = SegmentedVideoEncoder::init_with_timed_fragments( + self.base_path.clone(), + self.video_config, + encoder_config, + self.timed_fragments, + )?; if let Some(callback) = &self.disk_space_callback { encoder.set_disk_space_callback(callback.clone()); } @@ -682,8 +690,12 @@ impl MacOSFragmentedM4SCameraMuxer { prefer_videotoolbox_hw_input: false, }; - let mut encoder = - SegmentedVideoEncoder::init(self.base_path.clone(), self.video_config, encoder_config)?; + let mut encoder = SegmentedVideoEncoder::init_with_timed_fragments( + self.base_path.clone(), + self.video_config, + encoder_config, + true, + )?; if let Some(callback) = &self.disk_space_callback { encoder.set_disk_space_callback(callback.clone()); } diff --git a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs index 208ab9f4ea1..303c82ec61a 100644 --- a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs @@ -199,6 +199,7 @@ struct EncoderState { } pub struct WindowsFragmentedM4SMuxer { + timed_fragments: bool, base_path: PathBuf, video_config: VideoInfo, segment_duration: Duration, @@ -216,6 +217,7 @@ pub struct WindowsFragmentedM4SMuxer { } pub struct WindowsFragmentedM4SMuxerConfig { + pub timed_fragments: bool, pub segment_duration: Duration, pub preset: H264Preset, pub bpp: f32, @@ -229,6 +231,7 @@ pub struct WindowsFragmentedM4SMuxerConfig { impl Default for WindowsFragmentedM4SMuxerConfig { fn default() -> Self { Self { + timed_fragments: false, segment_duration: Duration::from_secs(2), preset: H264Preset::Ultrafast, bpp: H264EncoderBuilder::QUALITY_BPP, @@ -265,6 +268,7 @@ impl Muxer for WindowsFragmentedM4SMuxer { .unwrap_or_else(|| SharedPauseState::new(pause_flag)); let mut muxer = Self { + timed_fragments: config.timed_fragments, base_path: output_path, video_config, segment_duration: config.segment_duration, @@ -337,8 +341,12 @@ impl WindowsFragmentedM4SMuxer { prefer_videotoolbox_hw_input: false, }; - let mut encoder = - SegmentedVideoEncoder::init(self.base_path.clone(), self.video_config, encoder_config)?; + let mut encoder = SegmentedVideoEncoder::init_with_timed_fragments( + self.base_path.clone(), + self.video_config, + encoder_config, + self.timed_fragments, + )?; if let Some(callback) = &self.disk_space_callback { encoder.set_disk_space_callback(callback.clone()); } @@ -805,8 +813,12 @@ impl WindowsFragmentedM4SCameraMuxer { prefer_videotoolbox_hw_input: false, }; - let mut encoder = - SegmentedVideoEncoder::init(self.base_path.clone(), self.video_config, encoder_config)?; + let mut encoder = SegmentedVideoEncoder::init_with_timed_fragments( + self.base_path.clone(), + self.video_config, + encoder_config, + true, + )?; if let Some(callback) = &self.disk_space_callback { encoder.set_disk_space_callback(callback.clone()); } diff --git a/crates/recording/src/recovery.rs b/crates/recording/src/recovery.rs index 710684fbe16..3d1e70f67c9 100644 --- a/crates/recording/src/recovery.rs +++ b/crates/recording/src/recovery.rs @@ -5,7 +5,7 @@ use std::{ }; use cap_enc_ffmpeg::RelocatableSource; -use cap_enc_ffmpeg::fragmented_mp4::tail_is_complete; +use cap_enc_ffmpeg::fragmented_mp4::{read_fragment_prefix, tail_is_complete}; use cap_enc_ffmpeg::remux::{ concatenate_audio_to_ogg, concatenate_m4s_segments_with_init, concatenate_m4s_segments_with_init_validated, concatenate_video_fragments, get_media_duration, @@ -23,6 +23,7 @@ use tracing::{debug, warn}; use crate::output_pipeline::{HealthSender, PipelineHealthEvent, emit_health}; use crate::studio_recording::CompletedRecording; +mod prefix; mod preparing_observer; mod preparing_projection; #[cfg(unix)] @@ -381,7 +382,7 @@ impl RecoveryManager { .unwrap_or(0); let display_dir = segment_path.join("display"); - let display_info = Self::find_complete_fragments_with_init(&display_dir); + let display_info = Self::find_studio_fragments_with_init(&display_dir); let mut display_fragments = display_info.fragments; let mut display_init_segment = display_info.init_segment; @@ -403,7 +404,7 @@ impl RecoveryManager { let camera_dir = segment_path.join("camera"); let (camera_fragments, camera_init_segment) = { - let camera_info = Self::find_complete_fragments_with_init(&camera_dir); + let camera_info = Self::find_studio_fragments_with_init(&camera_dir); if camera_info.fragments.is_empty() { ( Self::probe_single_file(&segment_path.join("camera.mp4")).map(|p| vec![p]), @@ -456,6 +457,60 @@ impl RecoveryManager { }) } + fn find_studio_fragments_with_init(dir: &Path) -> FragmentsInfo { + let mut info = Self::find_complete_fragments_with_init(dir); + let Some(init) = info.init_segment.clone().or_else(|| { + let path = dir.join("init.mp4"); + path.is_file().then_some(path) + }) else { + return info; + }; + if let Ok(entries) = std::fs::read_dir(dir) { + for entry in entries.flatten() { + let path = entry.path(); + let Some(name) = path.file_name().and_then(|name| name.to_str()) else { + continue; + }; + let Some(final_name) = name.strip_suffix(".tmp") else { + continue; + }; + if Self::m4s_fragment_index(&dir.join(final_name)).is_none() + || dir.join(final_name).exists() + || !entry.file_type().is_ok_and(|kind| kind.is_file()) + { + continue; + } + // A malformed or empty tail has nothing to recover. Any other + // read error keeps the fragment listed: staged finalization + // refuses to publish a recording with a pending `.tmp`, so + // the recording stays retryable instead of losing it. + match read_fragment_prefix(&init, &path) { + Ok(_) => info.fragments.push(path), + Err(error) + if !matches!( + error.kind(), + std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof + ) => + { + warn!(path = %path.display(), %error, "Fragment prefix unreadable; keeping it for a retry"); + info.fragments.push(path); + } + Err(_) => {} + } + } + } + info.fragments.sort_by_key(|path| { + let path = if path.extension().is_some_and(|extension| extension == "tmp") { + path.with_extension("") + } else { + path.clone() + }; + Self::m4s_fragment_index(&path) + }); + info.init_segment = Some(init); + info + } + fn find_complete_fragments(dir: &Path) -> Vec { Self::find_complete_fragments_with_init(dir).fragments } @@ -638,6 +693,7 @@ impl RecoveryManager { fn collect_respawn_groups( dir: &Path, health_tx: Option<&HealthSender>, + recover_complete_prefixes: bool, ) -> Vec<(u32, PathBuf, Vec)> { let Ok(entries) = std::fs::read_dir(dir) else { return Vec::new(); @@ -667,7 +723,7 @@ impl RecoveryManager { continue; } - Self::rescue_pending_tmp_fragments(&respawn_dir, health_tx); + Self::rescue_pending_tmp_fragments(&respawn_dir, health_tx, recover_complete_prefixes); let Ok(dir_entries) = std::fs::read_dir(&respawn_dir) else { debug!( @@ -720,7 +776,25 @@ impl RecoveryManager { groups } - fn rescue_pending_tmp_fragments(dir: &Path, health_tx: Option<&HealthSender>) { + fn rescue_pending_tmp_fragments( + dir: &Path, + health_tx: Option<&HealthSender>, + recover_complete_prefixes: bool, + ) { + Self::rescue_pending_tmp_fragments_with_copy( + dir, + health_tx, + recover_complete_prefixes, + Self::copy_recovered_prefix, + ); + } + + fn rescue_pending_tmp_fragments_with_copy( + dir: &Path, + health_tx: Option<&HealthSender>, + recover_complete_prefixes: bool, + copy: impl Fn(&Path, &Path, u64) -> std::io::Result<()>, + ) { const MIN_VALID_TMP_SIZE: u64 = 100; let Ok(entries) = std::fs::read_dir(dir) else { @@ -742,7 +816,11 @@ impl RecoveryManager { continue; } let corrupt_marker = dir.join(format!("{name}.corrupt")); - if corrupt_marker.exists() { + if corrupt_marker.exists() + && (!recover_complete_prefixes + || std::fs::read_to_string(&corrupt_marker).ok().as_deref() + != Some("truncated_fragment")) + { continue; } let final_name = name.trim_end_matches(".tmp"); @@ -750,6 +828,37 @@ impl RecoveryManager { if final_path.exists() { continue; } + if recover_complete_prefixes && dir.join("init.mp4").is_file() { + match read_fragment_prefix(&dir.join("init.mp4"), &path) { + Ok(prefix) => match copy(&path, &final_path, prefix.length) { + Ok(()) => { + finalization_info!( + "Recovered complete fragment prefix from {}", + path.display() + ); + } + Err(error) => { + warn!(path = %path.display(), %error, "Could not copy recoverable fragment prefix"); + } + }, + Err(error) => { + if error.kind() == std::io::ErrorKind::InvalidData { + let _ = std::fs::write(&corrupt_marker, "truncated_fragment"); + } + warn!(path = %path.display(), %error, "Cannot recover fragment prefix"); + if let Some(health_tx) = health_tx { + emit_health( + health_tx, + PipelineHealthEvent::RecoveryFragmentCorrupt { + path: path.display().to_string(), + reason: error.to_string(), + }, + ); + } + } + } + continue; + } match tail_is_complete(&path) { Ok(true) => {} Ok(false) => { @@ -803,6 +912,39 @@ impl RecoveryManager { } } + fn copy_recovered_prefix( + source: &Path, + destination: &Path, + length: u64, + ) -> std::io::Result<()> { + use std::io::Read; + + let nonce = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(std::io::Error::other)? + .as_nanos(); + let temporary = + destination.with_extension(format!("recovery-{}-{nonce}", std::process::id())); + let result = (|| { + let source = std::fs::File::open(source)?; + let mut output = std::fs::OpenOptions::new() + .create_new(true) + .write(true) + .open(&temporary)?; + if std::io::copy(&mut source.take(length), &mut output)? != length { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "Fragment changed during recovery", + )); + } + output.sync_all()?; + drop(output); + prefix::publish(&temporary, destination) + })(); + let _ = std::fs::remove_file(&temporary); + result + } + fn is_video_file(path: &Path) -> bool { path.extension() .map(|e| e.eq_ignore_ascii_case("mp4") || e.eq_ignore_ascii_case("m4s")) @@ -940,7 +1082,16 @@ impl RecoveryManager { let mut total = Duration::ZERO; for fragment in fragments { - if let Some(duration) = get_media_duration(fragment) { + let prefix_duration = fragment + .extension() + .is_some_and(|extension| extension == "tmp") + .then(|| { + read_fragment_prefix(&fragment.with_file_name("init.mp4"), fragment) + .ok() + .map(|prefix| prefix.duration) + }) + .flatten(); + if let Some(duration) = prefix_duration.or_else(|| get_media_duration(fragment)) { total += duration; } } @@ -1121,9 +1272,30 @@ impl RecoveryManager { } } validate_recovery_manifests(&staged.join("content"))?; + let segments = staged.join("content/segments"); + for entry in std::fs::read_dir(segments)? { + let entry = entry?; + if entry.file_type()?.is_dir() { + for track in ["display", "camera"] { + let directory = entry.path().join(track); + Self::rescue_pending_tmp_fragments(&directory, None, true); + } + } + } let analyzing = Instant::now(); let staged_recording = Self::analyze_incomplete(&staged, &staged_meta) .ok_or(RecoveryError::NoRecoverableSegments)?; + if staged_recording.recoverable_segments.iter().any(|segment| { + segment + .display_fragments + .iter() + .chain(segment.camera_fragments.iter().flatten()) + .any(|path| path.extension().is_some_and(|extension| extension == "tmp")) + }) { + return Err(RecoveryError::Validation( + "Failed to stage a complete fragment prefix".into(), + )); + } debug!( elapsed_ms = analyzing.elapsed().as_secs_f64() * 1000.0, "Studio finalization staged analysis ready" @@ -1224,7 +1396,7 @@ impl RecoveryManager { segment.index )); if video_validation == VideoValidation::Full || segment.display_init_segment.is_none() { - for (_, init, fragments) in Self::collect_respawn_groups(&display_dir, None) { + for (_, init, fragments) in Self::collect_respawn_groups(&display_dir, None, true) { validate_recovery_video_inputs( &fragments, Some(&init), @@ -1379,6 +1551,7 @@ impl RecoveryManager { &display_output, None, video_validation, + true, ) .map(|_| ()) } else { @@ -1900,7 +2073,7 @@ impl RecoveryManager { "Missing required Instant audio".into(), )); } - Self::rescue_pending_tmp_fragments(&display, None); + Self::rescue_pending_tmp_fragments(&display, None, false); let video = Self::find_complete_fragments_with_init(&display); if video_validation == VideoValidation::Full || video.init_segment.is_none() { validate_recovery_video_inputs( @@ -1962,11 +2135,12 @@ impl RecoveryManager { output, None, video_validation, + false, )?; return Ok(None); } - Self::rescue_pending_tmp_fragments(audio_dir, None); + Self::rescue_pending_tmp_fragments(audio_dir, None, false); let audio_info = Self::find_complete_fragments_with_init(audio_dir); if audio_info.fragments.is_empty() { return Err(RecoveryError::Validation( @@ -2013,6 +2187,7 @@ impl RecoveryManager { &video_output, None, video_validation, + false, ); let audio = match audio { Some(handle) => handle.join().unwrap_or_else(|_| { @@ -2079,6 +2254,7 @@ impl RecoveryManager { output, health_tx, VideoValidation::Full, + true, ) } @@ -2087,15 +2263,17 @@ impl RecoveryManager { output: &Path, health_tx: Option<&HealthSender>, video_validation: VideoValidation, + recover_complete_prefixes: bool, ) -> Result { - Self::rescue_pending_tmp_fragments(fragmented_dir, health_tx); + Self::rescue_pending_tmp_fragments(fragmented_dir, health_tx, recover_complete_prefixes); let info = Self::find_complete_fragments_with_init(fragmented_dir); if info.fragments.is_empty() { return Err(RecoveryError::NoRecoverableSegments); } - let respawn_groups = Self::collect_respawn_groups(fragmented_dir, health_tx); + let respawn_groups = + Self::collect_respawn_groups(fragmented_dir, health_tx, recover_complete_prefixes); if respawn_groups.is_empty() { Self::finalize_fragments_to_progressive_mp4( @@ -6942,3 +7120,155 @@ mod transactional_recovery_tests { #[cfg(test)] mod preparing_canonical_probe; + +#[cfg(test)] +mod timed_fragment_tests { + use super::*; + + fn boxed(kind: &[u8; 4], values: &[u32]) -> Vec { + let mut bytes = (8 + values.len() as u32 * 4).to_be_bytes().to_vec(); + bytes.extend(kind); + bytes.extend(values.iter().flat_map(|value| value.to_be_bytes())); + bytes + } + + fn container(kind: &[u8; 4], children: &[Vec]) -> Vec { + let payload = children.concat(); + let mut bytes = (8 + payload.len() as u32).to_be_bytes().to_vec(); + bytes.extend(kind); + bytes.extend(payload); + bytes + } + + fn fixture(directory: &Path) -> (PathBuf, Vec) { + let track = container( + b"trak", + &[ + boxed(b"tkhd", &[0, 0, 0, 1]), + container(b"mdia", &[boxed(b"mdhd", &[0, 0, 0, 30])]), + ], + ); + let defaults = container(b"mvex", &[boxed(b"trex", &[0, 1, 1, 1, 4, 0])]); + std::fs::write( + directory.join("init.mp4"), + container(b"moov", &[track, defaults]), + ) + .unwrap(); + let moof = container( + b"moof", + &[container( + b"traf", + &[ + boxed(b"tfhd", &[0x20000, 1]), + boxed(b"tfdt", &[0, 0]), + boxed(b"trun", &[1, 60, 76]), + ], + )], + ); + assert_eq!(moof.len(), 68); + let mut complete = moof; + complete.extend(boxed(b"mdat", &[42; 60])); + let path = directory.join("segment_001.m4s.tmp"); + let mut partial = complete.clone(); + partial.extend(boxed(b"moof", &[0; 10])); + partial.extend(&boxed(b"mdat", &[0; 60])[..24]); + std::fs::write(&path, partial).unwrap(); + (path, complete) + } + + #[test] + fn discovers_pending_prefix_without_mutating_source() { + let directory = tempfile::tempdir().unwrap(); + let (source, _) = fixture(directory.path()); + let before = recovery_snapshot(directory.path()).unwrap(); + let info = RecoveryManager::find_studio_fragments_with_init(directory.path()); + assert_eq!(info.fragments, [source]); + assert_eq!(recovery_snapshot(directory.path()).unwrap(), before); + } + + #[cfg(unix)] + #[test] + fn unreadable_pending_fragment_stays_listed_for_a_retry() { + use std::os::unix::fs::PermissionsExt; + + let directory = tempfile::tempdir().unwrap(); + let (source, _) = fixture(directory.path()); + std::fs::set_permissions(&source, std::fs::Permissions::from_mode(0o000)).unwrap(); + if std::fs::File::open(&source).is_ok() { + // Permissions do not apply (running as root). + return; + } + let info = RecoveryManager::find_studio_fragments_with_init(directory.path()); + std::fs::set_permissions(&source, std::fs::Permissions::from_mode(0o644)).unwrap(); + assert_eq!(info.fragments, [source]); + } + + #[test] + fn studio_recovery_keeps_complete_prefix_and_original_bytes() { + let directory = tempfile::tempdir().unwrap(); + let (source, complete) = fixture(directory.path()); + let original = std::fs::read(&source).unwrap(); + let marker = directory.path().join("segment_001.m4s.tmp.corrupt"); + std::fs::write(&marker, "truncated_fragment").unwrap(); + RecoveryManager::rescue_pending_tmp_fragments(directory.path(), None, true); + let output = directory.path().join("segment_001.m4s"); + assert_eq!(std::fs::read(&output).unwrap(), complete); + assert_eq!(std::fs::read(&source).unwrap(), original); + RecoveryManager::rescue_pending_tmp_fragments(directory.path(), None, true); + assert_eq!(std::fs::read(&output).unwrap(), complete); + assert_eq!( + RecoveryManager::find_studio_fragments_with_init(directory.path()).fragments, + [output] + ); + } + + #[test] + fn instant_recovery_keeps_existing_partial_tail_policy() { + let directory = tempfile::tempdir().unwrap(); + let (source, _) = fixture(directory.path()); + let original = std::fs::read(&source).unwrap(); + RecoveryManager::rescue_pending_tmp_fragments(directory.path(), None, false); + assert!(!directory.path().join("segment_001.m4s").exists()); + assert_eq!(std::fs::read(source).unwrap(), original); + } + + #[test] + fn failed_prefix_copy_does_not_quarantine_valid_media() { + let directory = tempfile::tempdir().unwrap(); + let (source, complete) = fixture(directory.path()); + let original = std::fs::read(&source).unwrap(); + RecoveryManager::rescue_pending_tmp_fragments_with_copy( + directory.path(), + None, + true, + |_, _, _| Err(std::io::Error::other("injected storage failure")), + ); + assert!( + !directory + .path() + .join("segment_001.m4s.tmp.corrupt") + .exists() + ); + assert!(!directory.path().join("segment_001.m4s").exists()); + assert_eq!(std::fs::read(&source).unwrap(), original); + RecoveryManager::rescue_pending_tmp_fragments(directory.path(), None, true); + assert_eq!( + std::fs::read(directory.path().join("segment_001.m4s")).unwrap(), + complete + ); + } + + #[test] + fn prefix_copy_never_overwrites_an_existing_destination() { + let directory = tempfile::tempdir().unwrap(); + let (source, complete) = fixture(directory.path()); + let output = directory.path().join("segment_001.m4s"); + std::fs::write(&output, b"existing").unwrap(); + assert!( + RecoveryManager::copy_recovered_prefix(&source, &output, complete.len() as u64) + .is_err() + ); + assert_eq!(std::fs::read(output).unwrap(), b"existing"); + assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 3); + } +} diff --git a/crates/recording/src/recovery/prefix.rs b/crates/recording/src/recovery/prefix.rs new file mode 100644 index 00000000000..8a09e9ad604 --- /dev/null +++ b/crates/recording/src/recovery/prefix.rs @@ -0,0 +1,138 @@ +use std::{io, path::Path}; + +#[cfg(any(target_os = "macos", target_os = "linux"))] +pub(super) fn publish(source: &Path, destination: &Path) -> io::Result<()> { + use std::{ffi::CString, os::unix::ffi::OsStrExt}; + + let source_name = CString::new(source.as_os_str().as_bytes())?; + let destination_name = CString::new(destination.as_os_str().as_bytes())?; + #[cfg(target_os = "macos")] + let result = unsafe { + libc::renamex_np( + source_name.as_ptr(), + destination_name.as_ptr(), + libc::RENAME_EXCL, + ) + }; + #[cfg(target_os = "linux")] + let result = unsafe { + libc::renameat2( + libc::AT_FDCWD, + source_name.as_ptr(), + libc::AT_FDCWD, + destination_name.as_ptr(), + libc::RENAME_NOREPLACE, + ) + }; + if result == 0 { + Ok(()) + } else { + let error = io::Error::last_os_error(); + if error.kind() == io::ErrorKind::Unsupported + || matches!(error.raw_os_error(), Some(libc::ENOSYS | libc::EINVAL)) + { + // Production recovery publishes the enclosing workspace only after all prefixes validate. + copy_without_links(source, destination) + } else { + Err(error) + } + } +} + +#[cfg(windows)] +pub(super) fn publish(source: &Path, destination: &Path) -> io::Result<()> { + use std::os::windows::ffi::OsStrExt; + use windows::{Win32::Storage::FileSystem::MoveFileW, core::PCWSTR}; + + let source: Vec<_> = source.as_os_str().encode_wide().chain(Some(0)).collect(); + let destination: Vec<_> = destination + .as_os_str() + .encode_wide() + .chain(Some(0)) + .collect(); + if source[..source.len() - 1].contains(&0) || destination[..destination.len() - 1].contains(&0) + { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "Path contains a null character", + )); + } + unsafe { MoveFileW(PCWSTR(source.as_ptr()), PCWSTR(destination.as_ptr())) } + .map_err(|_| io::Error::last_os_error()) +} + +#[cfg(any(unix, test))] +fn copy_without_links(source: &Path, destination: &Path) -> io::Result<()> { + let mut input = std::fs::File::open(source)?; + let length = input.metadata()?.len(); + copy_to_new_file(&mut input, destination, length) +} + +#[cfg(any(unix, test))] +fn copy_to_new_file(input: &mut impl io::Read, destination: &Path, length: u64) -> io::Result<()> { + let mut output = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(destination)?; + let result = (|| { + if io::copy(input, &mut output)? != length { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "Incomplete recovered prefix copy", + )); + } + output.sync_all() + })(); + drop(output); + if result.is_err() { + let _ = std::fs::remove_file(destination); + } + result +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn publication_keeps_complete_bytes_without_replacing_existing_data() { + let directory = tempfile::tempdir().unwrap(); + let source = directory.path().join("temporary"); + let destination = directory.path().join("segment_001.m4s"); + std::fs::write(&source, b"complete prefix").unwrap(); + publish(&source, &destination).unwrap(); + if source.exists() { + assert_eq!(std::fs::read(&source).unwrap(), b"complete prefix"); + } + assert_eq!(std::fs::read(&destination).unwrap(), b"complete prefix"); + std::fs::write(&source, b"replacement").unwrap(); + assert!(publish(&source, &destination).is_err()); + assert_eq!(std::fs::read(&source).unwrap(), b"replacement"); + assert_eq!(std::fs::read(&destination).unwrap(), b"complete prefix"); + } + + #[test] + fn fallback_copy_preserves_source_and_refuses_existing_files() { + let directory = tempfile::tempdir().unwrap(); + let source = directory.path().join("temporary"); + let destination = directory.path().join("segment_001.m4s"); + std::fs::write(&source, b"complete prefix").unwrap(); + copy_without_links(&source, &destination).unwrap(); + assert_eq!(std::fs::read(&source).unwrap(), b"complete prefix"); + assert_eq!(std::fs::read(&destination).unwrap(), b"complete prefix"); + std::fs::write(&source, b"replacement").unwrap(); + assert!(copy_without_links(&source, &destination).is_err()); + assert_eq!(std::fs::read(&destination).unwrap(), b"complete prefix"); + } + + #[test] + fn failed_fallback_copy_removes_only_its_partial_file() { + let directory = tempfile::tempdir().unwrap(); + let destination = directory.path().join("segment_001.m4s"); + assert!(copy_to_new_file(&mut &b"short"[..], &destination, 100).is_err()); + assert!(!destination.exists()); + std::fs::write(&destination, b"existing").unwrap(); + assert!(copy_to_new_file(&mut &b"short"[..], &destination, 100).is_err()); + assert_eq!(std::fs::read(destination).unwrap(), b"existing"); + } +} diff --git a/crates/recording/src/studio_recording.rs b/crates/recording/src/studio_recording.rs index 762ae485795..0ba38ba565d 100644 --- a/crates/recording/src/studio_recording.rs +++ b/crates/recording/src/studio_recording.rs @@ -3319,6 +3319,7 @@ async fn create_segment_pipeline( builder .build::( crate::ffmpeg::SegmentedVideoMuxerConfig { + timed_fragments: true, segment_duration: Duration::from_secs(2), shared_pause_state: shared_pause_state.clone(), ..Default::default() @@ -3522,6 +3523,7 @@ async fn create_segment_pipeline( .with_start_gate(start_gate.clone()) .build::( crate::ffmpeg::SegmentedVideoMuxerConfig { + timed_fragments: true, segment_duration: Duration::from_secs(2), shared_pause_state: shared_pause_state.clone(), ..Default::default()