From 92e23ce2577c59e618980fc075841cc789bebcd0 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 13:11:10 +0900 Subject: [PATCH 1/3] fix(cap-muxer): finish the open fragment when the recorder dies --- Cargo.lock | 1 + crates/cap-muxer/Cargo.toml | 3 + crates/cap-muxer/src/main.rs | 20 +++-- crates/cap-muxer/tests/orphaned.rs | 139 +++++++++++++++++++++++++++++ 4 files changed, 157 insertions(+), 6 deletions(-) create mode 100644 crates/cap-muxer/tests/orphaned.rs diff --git a/Cargo.lock b/Cargo.lock index f059b7eff1e..b276a6e51b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1868,6 +1868,7 @@ dependencies = [ "cap-muxer-protocol", "ffmpeg-next", "libc", + "tempfile", "thiserror 1.0.69", "tracing", "tracing-subscriber", diff --git a/crates/cap-muxer/Cargo.toml b/crates/cap-muxer/Cargo.toml index 8549528cb9e..c58ca34b0d7 100644 --- a/crates/cap-muxer/Cargo.toml +++ b/crates/cap-muxer/Cargo.toml @@ -18,3 +18,6 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } [lints] workspace = true + +[dev-dependencies] +tempfile = "3" diff --git a/crates/cap-muxer/src/main.rs b/crates/cap-muxer/src/main.rs index ca5748e94b8..fc7092ff31b 100644 --- a/crates/cap-muxer/src/main.rs +++ b/crates/cap-muxer/src/main.rs @@ -25,37 +25,45 @@ fn main() -> ExitCode { match run() { Ok(()) => ExitCode::from(EXIT_OK), Err(MuxerError::Protocol(err)) => { - eprintln!("cap-muxer protocol error: {err}"); + report(format_args!("cap-muxer protocol error: {err}")); ExitCode::from(EXIT_PROTOCOL_ERROR) } Err(MuxerError::DiskFull(err)) => { - eprintln!("cap-muxer disk full: {err:#}"); + report(format_args!("cap-muxer disk full: {err:#}")); ExitCode::from(EXIT_DISK_FULL) } Err(MuxerError::FFmpeg(err)) => { - eprintln!("cap-muxer ffmpeg error: {err:#}"); + report(format_args!("cap-muxer ffmpeg error: {err:#}")); ExitCode::from(EXIT_FFMPEG_ERROR) } Err(MuxerError::Init(err)) => { - eprintln!("cap-muxer init error: {err:#}"); + report(format_args!("cap-muxer init error: {err:#}")); ExitCode::from(EXIT_INIT_ERROR) } Err(MuxerError::Abort(reason)) => { - eprintln!("cap-muxer parent abort: {reason}"); + report(format_args!("cap-muxer parent abort: {reason}")); ExitCode::from(EXIT_ABORT) } Err(MuxerError::BadState(reason)) => { - eprintln!("cap-muxer bad state: {reason}"); + report(format_args!("cap-muxer bad state: {reason}")); ExitCode::from(EXIT_BAD_STATE) } } } +/// Writes to stderr without panicking: when the recorder dies, the pipe it +/// read our stderr through is closed, and the open fragment must still be +/// finished after that. +fn report(message: std::fmt::Arguments) { + let _ = writeln!(io::stderr(), "{message}"); +} + fn init_tracing() { let filter = std::env::var("CAP_MUXER_LOG").unwrap_or_else(|_| "info".to_string()); let _ = tracing_subscriber::fmt() .with_env_filter(filter) .with_writer(std::io::stderr) + .log_internal_errors(false) .try_init(); } diff --git a/crates/cap-muxer/tests/orphaned.rs b/crates/cap-muxer/tests/orphaned.rs new file mode 100644 index 00000000000..f5f86bd79c2 --- /dev/null +++ b/crates/cap-muxer/tests/orphaned.rs @@ -0,0 +1,139 @@ +//! The recorder can die while cap-muxer runs. Its stdin then reaches EOF and +//! the pipe the recorder read cap-muxer's stderr through is closed; cap-muxer +//! must still finish the fragment it has open. + +use std::{ + io::Write, + process::{Command, Stdio}, +}; + +use cap_muxer_protocol::{ + Frame, InitVideo, PACKET_FLAG_KEYFRAME, Packet, STREAM_INDEX_VIDEO, StartParams, write_frame, + write_packet, +}; +use ffmpeg::{codec, format, frame}; + +const WIDTH: u32 = 320; +const HEIGHT: u32 = 240; +const FPS: i32 = 30; + +/// libx264 packets for `count` gray frames with a keyframe every `gop`. +fn encode(count: i64, gop: u32) -> (Vec, Vec<(i64, i64, bool, Vec)>) { + ffmpeg::init().unwrap(); + let codec = codec::encoder::find_by_name("libx264").expect("libx264"); + let mut context = codec::context::Context::new_with_codec(codec) + .encoder() + .video() + .unwrap(); + context.set_width(WIDTH); + context.set_height(HEIGHT); + context.set_format(format::Pixel::YUV420P); + context.set_time_base((1, FPS)); + context.set_frame_rate(Some((FPS, 1))); + context.set_gop(gop); + context.set_max_b_frames(0); + context.set_flags(codec::Flags::GLOBAL_HEADER); + let mut options = ffmpeg::Dictionary::new(); + options.set("preset", "ultrafast"); + options.set("tune", "zerolatency"); + let mut encoder = context.open_as_with(codec, options).unwrap(); + let extradata = unsafe { + let ctx = encoder.as_ptr(); + std::slice::from_raw_parts((*ctx).extradata, (*ctx).extradata_size as usize).to_vec() + }; + let mut packets = Vec::new(); + let mut packet = ffmpeg::Packet::empty(); + let mut drain = |encoder: &mut ffmpeg::encoder::Video, packets: &mut Vec<_>| { + while encoder.receive_packet(&mut packet).is_ok() { + packets.push(( + packet.pts().unwrap(), + packet.dts().unwrap(), + packet.is_key(), + packet.data().unwrap().to_vec(), + )); + } + }; + for index in 0..count { + let mut picture = frame::Video::new(format::Pixel::YUV420P, WIDTH, HEIGHT); + for plane in 0..3 { + picture.data_mut(plane).fill((index * 3 % 255) as u8); + } + picture.set_pts(Some(index)); + encoder.send_frame(&picture).unwrap(); + drain(&mut encoder, &mut packets); + } + encoder.send_eof().unwrap(); + drain(&mut encoder, &mut packets); + (extradata, packets) +} + +#[test] +fn an_orphaned_muxer_finishes_its_open_fragment() { + let (extradata, packets) = encode(75, 30); + let directory = tempfile::tempdir().unwrap(); + let output = directory.path().join("display"); + let mut child = Command::new(env!("CARGO_BIN_EXE_cap-muxer")) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let mut stdin = child.stdin.take().unwrap(); + write_frame( + &mut stdin, + &Frame::InitVideo(InitVideo { + codec: "libx264".into(), + width: WIDTH, + height: HEIGHT, + frame_rate_num: FPS, + frame_rate_den: 1, + time_base_num: 1, + time_base_den: FPS, + extradata, + segment_duration_ms: 1000, + }), + ) + .unwrap(); + write_frame( + &mut stdin, + &Frame::Start(StartParams { + output_directory: output.to_string_lossy().into_owned(), + init_segment_name: "init.mp4".into(), + media_segment_pattern: "segment_$Number%03d$.m4s".into(), + }), + ) + .unwrap(); + for (pts, dts, key, data) in &packets { + write_packet( + &mut stdin, + &Packet { + stream_index: STREAM_INDEX_VIDEO, + pts: *pts, + dts: *dts, + duration: 1, + flags: if *key { PACKET_FLAG_KEYFRAME } else { 0 }, + data, + }, + ) + .unwrap(); + } + stdin.flush().unwrap(); + // The recorder dies: its end of the stderr pipe closes, then stdin. + drop(child.stderr.take()); + drop(stdin); + let status = child.wait().unwrap(); + assert_eq!(status.code(), Some(0), "cap-muxer must finish, not panic"); + + let mut segments: Vec<_> = std::fs::read_dir(&output) + .unwrap() + .flatten() + .map(|entry| entry.file_name().to_string_lossy().into_owned()) + .filter(|name| name.starts_with("segment_")) + .collect(); + segments.sort(); + assert_eq!( + segments, + ["segment_001.m4s", "segment_002.m4s", "segment_003.m4s"], + "every fragment, including the one open when the recorder died, is finished" + ); +} From a9d29960108e899c4518bd450e8ec265e1b6cb8e Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 13:46:53 +0900 Subject: [PATCH 2/3] test(cap-muxer): name the encoded packet tuple --- crates/cap-muxer/tests/orphaned.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/cap-muxer/tests/orphaned.rs b/crates/cap-muxer/tests/orphaned.rs index f5f86bd79c2..9fa988ed304 100644 --- a/crates/cap-muxer/tests/orphaned.rs +++ b/crates/cap-muxer/tests/orphaned.rs @@ -17,8 +17,11 @@ const WIDTH: u32 = 320; const HEIGHT: u32 = 240; const FPS: i32 = 30; +/// pts, dts, keyframe, payload. +type EncodedPacket = (i64, i64, bool, Vec); + /// libx264 packets for `count` gray frames with a keyframe every `gop`. -fn encode(count: i64, gop: u32) -> (Vec, Vec<(i64, i64, bool, Vec)>) { +fn encode(count: i64, gop: u32) -> (Vec, Vec) { ffmpeg::init().unwrap(); let codec = codec::encoder::find_by_name("libx264").expect("libx264"); let mut context = codec::context::Context::new_with_codec(codec) From 109aa5e9a0bec969f25ee512514dbc0a50f1e3c7 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 14:30:00 +0900 Subject: [PATCH 3/3] test(cap-muxer): always log the EOF warning in the orphan test --- crates/cap-muxer/tests/orphaned.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/crates/cap-muxer/tests/orphaned.rs b/crates/cap-muxer/tests/orphaned.rs index 9fa988ed304..6980359da62 100644 --- a/crates/cap-muxer/tests/orphaned.rs +++ b/crates/cap-muxer/tests/orphaned.rs @@ -17,10 +17,8 @@ const WIDTH: u32 = 320; const HEIGHT: u32 = 240; const FPS: i32 = 30; -/// pts, dts, keyframe, payload. type EncodedPacket = (i64, i64, bool, Vec); -/// libx264 packets for `count` gray frames with a keyframe every `gop`. fn encode(count: i64, gop: u32) -> (Vec, Vec) { ffmpeg::init().unwrap(); let codec = codec::encoder::find_by_name("libx264").expect("libx264"); @@ -76,6 +74,7 @@ fn an_orphaned_muxer_finishes_its_open_fragment() { let directory = tempfile::tempdir().unwrap(); let output = directory.path().join("display"); let mut child = Command::new(env!("CARGO_BIN_EXE_cap-muxer")) + .env("CAP_MUXER_LOG", "warn") .stdin(Stdio::piped()) .stdout(Stdio::null()) .stderr(Stdio::piped())