diff --git a/crates/enc-ffmpeg/src/remux.rs b/crates/enc-ffmpeg/src/remux.rs index 1e791c40aba..0a93d609b0d 100644 --- a/crates/enc-ffmpeg/src/remux.rs +++ b/crates/enc-ffmpeg/src/remux.rs @@ -80,6 +80,76 @@ pub fn concatenate_video_fragments(fragments: &[PathBuf], output: &Path) -> Resu result } +/// Joins files cut from one recording timeline, such as the output of each +/// muxer process when one was replaced mid-recording. Each file keeps its own +/// start time, so video after a gap stays at its capture time instead of +/// closing the gap; the last frame before a gap is held across it. +pub fn concatenate_video_fragments_on_timeline( + fragments: &[PathBuf], + output: &Path, +) -> Result<(), RemuxError> { + if fragments.is_empty() { + return Err(RemuxError::NoFragments); + } + + for fragment in fragments { + if !fragment.exists() { + return Err(RemuxError::FragmentNotFound(fragment.clone())); + } + } + + let spans = fragments + .iter() + .map(|fragment| media_span(fragment)) + .collect::, _>>()?; + let concat_list_path = output.with_extension("concat.txt"); + let entries = fragments + .iter() + .enumerate() + .map(|(index, fragment)| { + let entry = concat_fragment_entry(fragment, &concat_list_path)?; + Ok(match timeline_duration(&spans, index) { + Some(duration) => format!("{entry}duration {duration}us\n"), + None => entry, + }) + }) + .collect::, RemuxError>>()?; + std::fs::write(&concat_list_path, entries.concat())?; + + let result = concatenate_with_concat_demuxer(&concat_list_path, output); + + let _ = std::fs::remove_file(&concat_list_path); + + result +} + +#[derive(Clone, Copy, Debug)] +struct MediaSpan { + start: i64, + duration: i64, +} + +fn media_span(path: &Path) -> Result, RemuxError> { + let input = ffmpeg::format::input(path)?; + let (start, duration) = (container_start_time(&input), input.duration()); + Ok((start != ffmpeg::ffi::AV_NOPTS_VALUE && duration > 0) + .then_some(MediaSpan { start, duration })) +} + +fn container_start_time(input: &avformat::context::Input) -> i64 { + unsafe { (*input.as_ptr()).start_time } +} + +/// How far into the joined timeline the file after `index` starts, in +/// microseconds: the gap between the two start times, but never less than the +/// file's own duration, so overlapping files still play end to end. `None` +/// leaves the concat demuxer to use the file's probed duration. +fn timeline_duration(spans: &[Option], index: usize) -> Option { + let current = spans.get(index).copied().flatten()?; + let next = spans.get(index + 1).copied().flatten()?; + Some((next.start - current.start).max(current.duration)) +} + fn concat_fragment_entry(fragment: &Path, concat_list: &Path) -> Result { let fragment = std::path::absolute(fragment)?; let concat_list = std::path::absolute(concat_list)?; @@ -1374,7 +1444,8 @@ mod tests { concatenate_audio_to_ogg, concatenate_m4s_segments_with_init, concatenate_m4s_segments_with_init_validated, concatenate_m4s_segments_with_init_validated_with_reader, concatenate_video_fragments, - get_video_capture_fps, get_video_fps, probe_video_pts_ladder, write_concat_list, + concatenate_video_fragments_on_timeline, get_video_capture_fps, get_video_fps, + probe_video_pts_ladder, write_concat_list, }; use std::{ path::{Path, PathBuf}, @@ -1794,6 +1865,113 @@ mod tests { assert!(!output.with_extension("concat.txt").exists()); } + fn first_video_pts_after(path: &Path, frames_before: usize) -> f64 { + let (time_base, packets) = video_packet_contents(path); + let mut pts: Vec = packets.iter().filter_map(|packet| packet.pts).collect(); + pts.sort_unstable(); + pts[frames_before] as f64 * f64::from(time_base) + } + + fn timeline_inputs(parent: &Path, second_start_frame: u64) -> [PathBuf; 2] { + let frames = |range: std::ops::Range| -> Vec { + range + .map(|frame| Duration::from_nanos(frame * 1_000_000_000 / 30)) + .collect() + }; + let first_dir = parent.join("first"); + let second_dir = parent.join("second"); + std::fs::create_dir_all(&first_dir).unwrap(); + std::fs::create_dir_all(&second_dir).unwrap(); + let second = second_dir.join("shifted.mp4"); + shift_test_mp4( + &encode_test_mp4(&second_dir, &frames(0..12)), + &second, + second_start_frame as f64 / 30.0, + ); + [encode_test_mp4(&first_dir, &frames(0..12)), second] + } + + /// A copy of `input` whose packets start `offset` seconds later, like a + /// replacement muxer's output on the recording's shared timeline. + fn shift_test_mp4(input: &Path, output: &Path, offset: f64) { + let mut ictx = ffmpeg::format::input(input).unwrap(); + let mut octx = ffmpeg::format::output(output).unwrap(); + let stream = ictx.streams().best(ffmpeg::media::Type::Video).unwrap(); + let (index, time_base) = (stream.index(), stream.time_base()); + let mut output_stream = octx.add_stream(None).unwrap(); + output_stream.set_parameters(stream.parameters()); + unsafe { + (*output_stream.as_mut_ptr()) + .codecpar + .as_mut() + .unwrap() + .codec_tag = 0; + } + octx.write_header().unwrap(); + let output_time_base = octx.stream(0).unwrap().time_base(); + let shift = (offset * f64::from(time_base.denominator()) / f64::from(time_base.numerator())) + .round() as i64; + for (stream, mut packet) in ictx.packets() { + if stream.index() != index { + continue; + } + packet.set_pts(packet.pts().map(|pts| pts + shift)); + packet.set_dts(packet.dts().map(|dts| dts + shift)); + packet.rescale_ts(time_base, output_time_base); + packet.set_stream(0); + packet.set_position(-1); + packet.write_interleaved(&mut octx).unwrap(); + } + octx.write_trailer().unwrap(); + } + + #[test] + fn timeline_concat_keeps_a_later_file_at_its_start_time() { + let directory = tempfile::tempdir().unwrap(); + let inputs = timeline_inputs(directory.path(), 30); + let start = |path: &Path| { + super::container_start_time(&ffmpeg::format::input(path).unwrap()) as f64 + / f64::from(ffmpeg::ffi::AV_TIME_BASE) + }; + assert!( + (start(&inputs[1]) - 1.0).abs() < 0.002, + "{}", + start(&inputs[1]) + ); + + let joined = directory.path().join("timeline.mp4"); + concatenate_video_fragments_on_timeline(&inputs, &joined).unwrap(); + let second_at = first_video_pts_after(&joined, 12); + assert!( + (second_at - 1.0).abs() < 0.002, + "second file starts at {second_at}" + ); + assert_eq!(video_packet_contents(&joined).1.len(), 24); + assert!(!joined.with_extension("concat.txt").exists()); + + let end_to_end = directory.path().join("end-to-end.mp4"); + concatenate_video_fragments(&inputs, &end_to_end).unwrap(); + assert!(first_video_pts_after(&end_to_end, 12) < 0.5); + } + + #[test] + fn timeline_concat_plays_overlapping_files_end_to_end() { + let directory = tempfile::tempdir().unwrap(); + let inputs = timeline_inputs(directory.path(), 6); + + let joined = directory.path().join("timeline.mp4"); + concatenate_video_fragments_on_timeline(&inputs, &joined).unwrap(); + let packets = video_packet_contents(&joined).1; + assert_eq!(packets.len(), 24); + let dts: Vec = packets.iter().filter_map(|packet| packet.dts).collect(); + assert!(dts.windows(2).all(|pair| pair[0] < pair[1]), "{dts:?}"); + let second_at = first_video_pts_after(&joined, 12); + assert!( + second_at >= 0.39, + "overlap played over the first file: {second_at}" + ); + } + #[test] fn concat_audio_and_video_with_quoted_parent_directories() { assert_concat_roundtrip("recording 'quoted' λ"); diff --git a/crates/recording/src/output_pipeline/oop_muxer.rs b/crates/recording/src/output_pipeline/oop_muxer.rs index 1cacf4fe2d8..909c6321f98 100644 --- a/crates/recording/src/output_pipeline/oop_muxer.rs +++ b/crates/recording/src/output_pipeline/oop_muxer.rs @@ -365,7 +365,9 @@ impl MuxerSubprocess { data, }; - match write_packet_frame(stdin, &packet) { + // Flush every packet: anything left in the buffer is lost if the + // recorder dies, and a dead child only shows up as a write error. + match write_packet_frame(stdin, &packet).and_then(|()| Ok(stdin.flush()?)) { Ok(()) => { self.packets_written.fetch_add(1, Ordering::Relaxed); Ok(()) @@ -871,6 +873,61 @@ mod tests { (directory, subprocess) } + fn script_subprocess(script: &str) -> (tempfile::TempDir, MuxerSubprocess, PathBuf) { + let directory = tempfile::tempdir().unwrap(); + let binary = directory.path().join("muxer"); + std::fs::write(&binary, format!("#!/bin/sh\n{script}\n")).unwrap(); + std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o700)).unwrap(); + let config = MuxerSubprocessConfig { + output_directory: directory.path().join("video"), + init_segment_name: "init.mp4".into(), + media_segment_pattern: "segment_$Number%03d$.m4s".into(), + video_init: None, + audio_init: None, + }; + let received = directory.path().join("received"); + let subprocess = MuxerSubprocess::spawn(binary, config, None).unwrap(); + (directory, subprocess, received) + } + + #[test] + fn every_packet_reaches_the_child_before_the_next_one_is_written() { + let (directory, mut subprocess, received) = + script_subprocess("exec cat > \"$(dirname \"$0\")/received\""); + let size = |path: &Path| std::fs::metadata(path).map(|m| m.len()).unwrap_or(0); + let deadline = Instant::now() + Duration::from_secs(5); + while size(&received) == 0 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(5)); + } + let after_start = size(&received); + assert!(after_start > 0, "init and start frames never arrived"); + subprocess + .write_video_packet(0, 0, 1, true, &[7; 1000]) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + while size(&received) < after_start + 1000 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(5)); + } + assert!( + size(&received) >= after_start + 1000, + "a written packet stayed in the recorder's buffer" + ); + drop(subprocess); + drop(directory); + } + + #[test] + fn a_dead_child_is_reported_on_the_next_packet() { + let (_directory, mut subprocess, _) = script_subprocess("sleep 0.3; exit 3"); + let child = subprocess.child.as_mut().unwrap(); + let status = child.wait().unwrap(); + assert_eq!(status.code(), Some(3)); + assert!(matches!( + subprocess.write_video_packet(0, 0, 1, true, &[7; 1000]), + Err(MuxerSubprocessError::Crashed(_)) + )); + } + #[test] fn disk_full_stays_terminal_and_finish_confirms_shutdown() { let (_directory, mut subprocess) = exited_subprocess(EXIT_DISK_FULL, 2); diff --git a/crates/recording/src/recovery.rs b/crates/recording/src/recovery.rs index 1926978a9f3..3ae062fb383 100644 --- a/crates/recording/src/recovery.rs +++ b/crates/recording/src/recovery.rs @@ -8,9 +8,10 @@ use cap_enc_ffmpeg::RelocatableSource; 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, - get_video_base_fps, merge_video_audio, probe_media_valid, probe_video_can_decode, - probe_video_seek_points, remux_file, + concatenate_m4s_segments_with_init_validated, concatenate_video_fragments, + concatenate_video_fragments_on_timeline, get_media_duration, get_video_base_fps, + merge_video_audio, probe_media_valid, probe_video_can_decode, probe_video_seek_points, + remux_file, }; use cap_project::{ AudioMeta, Cursors, MultipleSegment, MultipleSegments, ProjectConfiguration, RecordingMeta, @@ -2336,7 +2337,8 @@ impl RecoveryManager { std::fs::rename(&group_outputs[0], output) .map_err(|e| RecoveryError::VideoConcat(cap_enc_ffmpeg::remux::RemuxError::Io(e))) } else { - concatenate_video_fragments(&group_outputs, output).map_err(RecoveryError::VideoConcat) + concatenate_video_fragments_on_timeline(&group_outputs, output) + .map_err(RecoveryError::VideoConcat) }; for tmp in &temp_paths { diff --git a/crates/recording/tests/oop_muxer.rs b/crates/recording/tests/oop_muxer.rs index c705d407709..bd7b889bfbc 100644 --- a/crates/recording/tests/oop_muxer.rs +++ b/crates/recording/tests/oop_muxer.rs @@ -217,6 +217,43 @@ fn subprocess_exits_after_finish_or_abort_without_waiting_for_stdin_eof() { } } +#[cfg(unix)] +#[test] +fn finish_waits_for_a_muxer_that_takes_more_than_five_seconds_to_exit() { + use cap_muxer_protocol::{Frame, read_frame}; + use std::io::Cursor; + use std::os::unix::fs::PermissionsExt; + use std::time::{Duration, Instant}; + + let directory = TempDir::new().unwrap(); + let binary = directory.path().join("slow-exit-muxer"); + std::fs::write(&binary, "#!/bin/sh\ncat > \"$0.received\"\nsleep 6\n").unwrap(); + std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o700)).unwrap(); + let config = minimal_video_config(&directory.path().join("video"), Vec::new()); + let mut subprocess = MuxerSubprocess::spawn(binary, config, None).unwrap(); + let payload = vec![42; 128 * 1024]; + subprocess + .write_video_packet(0, 0, 1, true, &payload) + .unwrap(); + let started = Instant::now(); + let report = subprocess.finish().unwrap(); + assert_eq!(report.exit_code, Some(0)); + assert!(started.elapsed() >= Duration::from_secs(5)); + + let received = std::fs::read(directory.path().join("slow-exit-muxer.received")).unwrap(); + let mut reader = Cursor::new(received.as_slice()); + assert!(matches!( + read_frame(&mut reader).unwrap(), + Frame::InitVideo(_) + )); + assert!(matches!(read_frame(&mut reader).unwrap(), Frame::Start(_))); + let Frame::Packet(packet) = read_frame(&mut reader).unwrap() else { + panic!("expected the video packet"); + }; + assert_eq!(packet.data, payload); + assert!(matches!(read_frame(&mut reader).unwrap(), Frame::Finish)); +} + #[cfg(unix)] #[test] fn finish_preserves_packets_when_a_slow_muxer_takes_more_than_five_seconds() { @@ -232,10 +269,10 @@ fn finish_preserves_packets_when_a_slow_muxer_takes_more_than_five_seconds() { let config = minimal_video_config(&directory.path().join("video"), Vec::new()); let mut subprocess = MuxerSubprocess::spawn(binary, config, None).unwrap(); let payload = vec![42; 128 * 1024]; + let started = Instant::now(); subprocess .write_video_packet(0, 0, 1, true, &payload) .unwrap(); - let started = Instant::now(); let report = subprocess.finish().unwrap(); assert_eq!(report.exit_code, Some(0)); assert!(started.elapsed() >= Duration::from_secs(5)); diff --git a/crates/verify/expectations.toml b/crates/verify/expectations.toml index 89ac5bfb1a3..42f77dc3e30 100644 --- a/crates/verify/expectations.toml +++ b/crates/verify/expectations.toml @@ -395,51 +395,3 @@ class = "missing" bound = { count = 1 } reason = "NEW (bug B1, see loss-time): no fragment of the static recording is complete when the recorder dies, so recovery has nothing (no frames)" link = "https://github.com/CapSoftware/Cap/pull/2641" - -[[xfail]] -case = "crash/recorder-kill/oop/r*" -class = "loss-time" -platforms = ["macos"] -bound = { count = 1, secs = [2.5, 10] } -reason = "NEW (bug C): the out-of-process muxer writes packets into a 1 MiB BufWriter around cap-muxer's stdin and flushes only at Start and Finish (oop_muxer.rs MuxerSubprocess::write_packet), so encoded video stays in the recorder's memory: killed after 4 to 8 s, nothing reached cap-muxer (init.mp4 0 bytes, no segment). Bound: everything handed before the kill, at most 10 s" -link = "https://github.com/CapSoftware/Cap/pull/2643" - -[[xfail]] -case = "crash/recorder-kill/oop/r*" -class = "missing" -platforms = ["macos"] -bound = { count = 1 } -reason = "NEW (bug C, see loss-time): recovery finds no segment at all (no frames)" -link = "https://github.com/CapSoftware/Cap/pull/2643" - -[[xfail]] -case = "crash/recorder-kill/oop/r*" -class = "loss" -platforms = ["macos"] -bound = { count = 1, frames = [64, 300] } -reason = "NEW (bug C, see loss-time): every handed frame is lost, more than one fragment plus the frames in flight (at most the 10 s handed at 30 fps)" -link = "https://github.com/CapSoftware/Cap/pull/2643" - -[[xfail]] -case = "crash/muxer-kill/oop" -class = "error" -platforms = ["macos"] -bound = { count = 1 } -reason = "NEW (bug C): after cap-muxer is killed the recorder keeps buffering packets for the dead child (no flush, so no write error and no respawn), and stop() fails the display track: \"subprocess was reaped with unusable output\"" -link = "https://github.com/CapSoftware/Cap/pull/2643" - -[[xfail]] -case = "crash/recorder-kill/oop/static-screen" -class = "loss-time" -platforms = ["macos"] -bound = { count = 1, secs = [2.5, 20.5] } -reason = "NEW (bugs B1 and C): the static recording's fragments span 20 s of capture (frame-count keyframe interval, see the in-process entry) and the out-of-process muxer had not received any packet (1 MiB unflushed buffer)" -link = "https://github.com/CapSoftware/Cap/pull/2643" - -[[xfail]] -case = "crash/recorder-kill/oop/static-screen" -class = "missing" -platforms = ["macos"] -bound = { count = 1 } -reason = "NEW (bugs B1 and C, see loss-time): nothing recoverable (no frames)" -link = "https://github.com/CapSoftware/Cap/pull/2643"