From 1aab3b5066f2ef1954c9c33defdc4644e054a19b Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:31:01 +0900 Subject: [PATCH 1/4] feat(enc-ffmpeg): let encoders write packets to any packet sink --- crates/enc-ffmpeg/src/audio/aac.rs | 4 ++++ crates/enc-ffmpeg/src/audio/opus.rs | 13 +++++++++++ crates/enc-ffmpeg/src/video/h264.rs | 35 +++++++++++++++++++++++++++-- 3 files changed, 50 insertions(+), 2 deletions(-) diff --git a/crates/enc-ffmpeg/src/audio/aac.rs b/crates/enc-ffmpeg/src/audio/aac.rs index 89ce84fcd3a..c1fb74a6e62 100644 --- a/crates/enc-ffmpeg/src/audio/aac.rs +++ b/crates/enc-ffmpeg/src/audio/aac.rs @@ -123,6 +123,10 @@ impl AACEncoder { ) -> Result<(), ffmpeg::Error> { self.base.send_frame(frame, timestamp, sink) } + + pub fn flush_to(&mut self, sink: &mut impl PacketSink) -> Result<(), ffmpeg::Error> { + self.base.flush(sink) + } } impl AudioEncoder for AACEncoder { diff --git a/crates/enc-ffmpeg/src/audio/opus.rs b/crates/enc-ffmpeg/src/audio/opus.rs index 3bcfb974cff..5182c29687b 100644 --- a/crates/enc-ffmpeg/src/audio/opus.rs +++ b/crates/enc-ffmpeg/src/audio/opus.rs @@ -106,6 +106,19 @@ impl OpusEncoder { pub fn flush(&mut self, output: &mut format::context::Output) -> Result<(), ffmpeg::Error> { self.base.flush(output) } + + pub fn queue_frame_to( + &mut self, + frame: frame::Audio, + timestamp: Duration, + sink: &mut impl crate::PacketSink, + ) -> Result<(), ffmpeg::Error> { + self.base.send_frame(frame, timestamp, sink) + } + + pub fn flush_to(&mut self, sink: &mut impl crate::PacketSink) -> Result<(), ffmpeg::Error> { + self.base.flush(sink) + } } fn select_output_rate(input_rate: i32, supported_rates: &[i32]) -> Option { diff --git a/crates/enc-ffmpeg/src/video/h264.rs b/crates/enc-ffmpeg/src/video/h264.rs index 6ad7b80bf87..446a2f4cd90 100644 --- a/crates/enc-ffmpeg/src/video/h264.rs +++ b/crates/enc-ffmpeg/src/video/h264.rs @@ -11,7 +11,7 @@ use ffmpeg::{ }; use tracing::{debug, error, info, trace, warn}; -use crate::base::EncoderBase; +use crate::base::{EncoderBase, PacketSink}; use crate::video::h264_packet::H264PacketEncoder; #[cfg(target_os = "macos")] use crate::video::videotoolbox_hw::VideoToolboxHwFrames; @@ -924,9 +924,18 @@ impl H264Encoder { pub fn queue_frame( &mut self, - mut frame: frame::Video, + frame: frame::Video, timestamp: Duration, output: &mut format::context::Output, + ) -> Result<(), QueueFrameError> { + self.queue_frame_to(frame, timestamp, output) + } + + pub fn queue_frame_to( + &mut self, + mut frame: frame::Video, + timestamp: Duration, + output: &mut S, ) -> Result<(), QueueFrameError> { self.base .update_pts(&mut frame, timestamp, &mut self.encoder); @@ -1010,6 +1019,28 @@ impl H264Encoder { self.base.process_eof(output, &mut self.encoder) } + pub fn flush_to( + &mut self, + output: &mut S, + ) -> Result<(), ffmpeg::Error> { + self.base.process_eof(output, &mut self.encoder) + } + + /// Configured target bitrate in bits per second, estimated from the + /// output geometry when the encoder runs without a bitrate target. + pub fn bit_rate(&self) -> u64 { + let configured = unsafe { (*self.encoder.as_ptr()).bit_rate }; + if configured > 0 { + return configured as u64; + } + get_bitrate( + self.output_width, + self.output_height, + 30.0, + H264EncoderBuilder::QUALITY_BPP, + ) as u64 + } + pub fn set_packet_stats(&mut self, stats: Arc) { self.base.set_packet_stats(stats); } From 5ed06562be060cfa35a594db291901c5695e4814 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:31:01 +0900 Subject: [PATCH 2/4] feat(enc-ffmpeg): add bounded write-behind muxing for segmented and audio outputs --- crates/enc-ffmpeg/src/mux/dash_audio.rs | 199 +++- crates/enc-ffmpeg/src/mux/fragmented_audio.rs | 97 +- crates/enc-ffmpeg/src/mux/mod.rs | 1 + crates/enc-ffmpeg/src/mux/ogg.rs | 70 +- crates/enc-ffmpeg/src/mux/segmented_stream.rs | 392 +++++--- crates/enc-ffmpeg/src/mux/write_behind.rs | 892 ++++++++++++++++++ 6 files changed, 1483 insertions(+), 168 deletions(-) create mode 100644 crates/enc-ffmpeg/src/mux/write_behind.rs diff --git a/crates/enc-ffmpeg/src/mux/dash_audio.rs b/crates/enc-ffmpeg/src/mux/dash_audio.rs index b3dfcc0079e..83bfb4f3036 100644 --- a/crates/enc-ffmpeg/src/mux/dash_audio.rs +++ b/crates/enc-ffmpeg/src/mux/dash_audio.rs @@ -1,6 +1,7 @@ use super::fragment_metadata::read_fragment_metadata; use crate::audio::aac::{AACEncoder, AACEncoderError}; use crate::mux::segmented_stream::{SegmentCompletedEvent, SegmentMediaType}; +use crate::mux::write_behind::{AUDIO_BUDGET_BYTES, MuxSink, MuxTarget}; use cap_media_info::AudioInfo; use ffmpeg::{format, frame}; use serde::Serialize; @@ -47,6 +48,11 @@ fn sync_file(path: &Path) { pub struct DashAudioSegmentEncoder { base_path: PathBuf, encoder: AACEncoder, + mux: MuxSink, +} + +struct AudioSegmentMux { + base_path: PathBuf, output: format::context::Output, initial_padding: Duration, @@ -60,6 +66,26 @@ pub struct DashAudioSegmentEncoder { init_notified: bool, } +impl MuxTarget for AudioSegmentMux { + fn output(&self) -> &format::context::Output { + &self.output + } + + fn output_mut(&mut self) -> &mut format::context::Output { + &mut self.output + } + + fn packet_written(&mut self) { + self.try_notify_init_segment(); + + self.frames_since_segment_scan += 1; + if self.frames_since_segment_scan >= 10 { + self.frames_since_segment_scan = 0; + self.flush_pending_segments(); + } + } +} + #[derive(Debug, Clone)] pub struct AudioSegmentInfo { pub path: PathBuf, @@ -197,9 +223,8 @@ impl DashAudioSegmentEncoder { "Initialized DASH audio segment encoder (init.mp4 + m4s segments)" ); - let instance = Self { - base_path, - encoder, + let mux = AudioSegmentMux { + base_path: base_path.clone(), output, initial_padding, current_index: 1, @@ -209,21 +234,83 @@ impl DashAudioSegmentEncoder { init_notified: false, }; - instance.write_in_progress_manifest(); + mux.write_in_progress_manifest(); + + Ok(Self { + base_path, + encoder, + mux: MuxSink::Direct(mux), + }) + } + + /// Moves muxing, segment publication and manifest writes onto a + /// dedicated writer thread so a stalled output volume no longer blocks + /// the caller. Bytes on disk and their order are unchanged. + pub fn enable_write_behind(&mut self) -> std::io::Result<()> { + self.mux + .enable_write_behind("audio-segments", AUDIO_BUDGET_BYTES) + } - Ok(instance) + pub fn is_write_behind(&self) -> bool { + self.mux.is_write_behind() } pub fn set_segment_callback(&mut self, tx: std::sync::mpsc::Sender) { - self.segment_tx = Some(tx); - self.try_notify_init_segment(); + self.mux.with_target(|mux| { + mux.segment_tx = Some(tx); + mux.try_notify_init_segment(); + }); } + pub fn queue_frame( + &mut self, + frame: frame::Audio, + timestamp: Duration, + ) -> Result<(), QueueFrameError> { + self.encoder + .send_frame_to(frame, timestamp, &mut self.mux)?; + + self.mux.frame_queued(); + + Ok(()) + } + + pub fn finish(&mut self) -> Result<(), FinishError> { + let encoder = &mut self.encoder; + let flush_result = self + .mux + .finish_flush(|mux| encoder.flush_to(mux)) + .map_err(std::io::Error::from)? + .inspect_err(|error| { + tracing::warn!(%error, "Audio encoder flush failed"); + }); + self.mux.with_target(|mux| mux.finish(flush_result)) + } + + pub fn finish_with_timestamp(&mut self, _timestamp: Duration) -> Result<(), FinishError> { + self.finish() + } + + pub fn completed_segments(&self) -> Vec { + self.mux + .with_target_ref(|mux| mux.completed_segments.clone()) + } + + pub fn base_path(&self) -> &Path { + &self.base_path + } + + pub fn init_segment_path(&self) -> PathBuf { + self.base_path.join(INIT_SEGMENT_NAME) + } +} + +impl AudioSegmentMux { fn try_notify_init_segment(&mut self) { if self.init_notified { return; } - let init_path = self.init_segment_path(); + let init_path = self.base_path.join(INIT_SEGMENT_NAME); if let Ok(meta) = std::fs::metadata(&init_path) && meta.len() > 0 { @@ -247,25 +334,6 @@ impl DashAudioSegmentEncoder { } } - pub fn queue_frame( - &mut self, - frame: frame::Audio, - timestamp: Duration, - ) -> Result<(), QueueFrameError> { - self.encoder - .send_frame(frame, timestamp, &mut self.output)?; - - self.try_notify_init_segment(); - - self.frames_since_segment_scan += 1; - if self.frames_since_segment_scan >= 10 { - self.frames_since_segment_scan = 0; - self.flush_pending_segments(); - } - - Ok(()) - } - fn flush_pending_segments(&mut self) { let first_index = self.current_index; loop { @@ -367,10 +435,7 @@ impl DashAudioSegmentEncoder { } } - pub fn finish(&mut self) -> Result<(), FinishError> { - let flush_result = self.encoder.flush(&mut self.output).inspect_err(|error| { - tracing::warn!(%error, "Audio encoder flush failed"); - }); + fn finish(&mut self, flush_result: Result<(), ffmpeg::Error>) -> Result<(), FinishError> { let trailer_result = self.output.write_trailer().inspect_err(|error| { tracing::warn!(%error, "Audio trailer publication failed"); }); @@ -394,10 +459,6 @@ impl DashAudioSegmentEncoder { Ok(()) } - pub fn finish_with_timestamp(&mut self, _timestamp: Duration) -> Result<(), FinishError> { - self.finish() - } - fn finalize_pending_tmp_files(&self) { let Ok(entries) = std::fs::read_dir(&self.base_path) else { return; @@ -523,18 +584,6 @@ impl DashAudioSegmentEncoder { let manifest_path = self.base_path.join("manifest.json"); atomic_write_json(&manifest_path, &manifest) } - - pub fn completed_segments(&self) -> &[AudioSegmentInfo] { - &self.completed_segments - } - - pub fn base_path(&self) -> &Path { - &self.base_path - } - - pub fn init_segment_path(&self) -> PathBuf { - self.base_path.join(INIT_SEGMENT_NAME) - } } #[cfg(test)] @@ -570,6 +619,62 @@ mod tests { frame } + fn record_audio_segments(base_path: &Path, write_behind: bool) -> Vec<(u32, u64, bool)> { + let mut encoder = DashAudioSegmentEncoder::init( + base_path.to_path_buf(), + test_audio_info(), + DashAudioSegmentEncoderConfig::default(), + ) + .unwrap(); + if write_behind { + encoder.enable_write_behind().unwrap(); + } + let (tx, rx) = mpsc::channel(); + encoder.set_segment_callback(tx); + for i in 0..500u64 { + encoder + .queue_frame( + create_test_audio_frame(1024, i * 1024), + Duration::from_secs_f64(i as f64 * 1024.0 / 48_000.0), + ) + .unwrap(); + } + encoder.finish().unwrap(); + drop(encoder); + rx.try_iter() + .map(|event| (event.index, event.file_size, event.is_init)) + .collect() + } + + #[test] + fn write_behind_publishes_the_same_audio_segments_as_inline_muxing() { + ffmpeg::init().unwrap(); + let temp = tempfile::tempdir().unwrap(); + let inline = temp.path().join("inline"); + let behind = temp.path().join("behind"); + + let inline_events = record_audio_segments(&inline, false); + let behind_events = record_audio_segments(&behind, true); + + assert!(inline_events.len() >= 3); + assert_eq!(inline_events, behind_events); + let files = |path: &Path| { + let mut files: Vec<(String, Vec)> = std::fs::read_dir(path) + .unwrap() + .flatten() + .map(|entry| { + ( + entry.file_name().to_string_lossy().into_owned(), + std::fs::read(entry.path()).unwrap(), + ) + }) + .collect(); + files.sort(); + files + }; + assert_eq!(files(&inline), files(&behind)); + } + #[test] fn audio_manifest_and_events_match_muxer_timeline() { ffmpeg::init().unwrap(); diff --git a/crates/enc-ffmpeg/src/mux/fragmented_audio.rs b/crates/enc-ffmpeg/src/mux/fragmented_audio.rs index 1203ad72ce4..a6d575f2930 100644 --- a/crates/enc-ffmpeg/src/mux/fragmented_audio.rs +++ b/crates/enc-ffmpeg/src/mux/fragmented_audio.rs @@ -2,15 +2,28 @@ use cap_media_info::AudioInfo; use ffmpeg::{format, frame}; use std::{path::PathBuf, time::Duration}; +use super::write_behind::{AUDIO_BUDGET_BYTES, MuxSink, MuxTarget}; use crate::audio::aac::{AACEncoder, AACEncoderError}; pub struct FragmentedAudioFile { encoder: AACEncoder, - output: format::context::Output, + output: MuxSink, finished: bool, has_frames: bool, } +struct AudioOutput(format::context::Output); + +impl MuxTarget for AudioOutput { + fn output(&self) -> &format::context::Output { + &self.0 + } + + fn output_mut(&mut self) -> &mut format::context::Output { + &mut self.0 + } +} + #[derive(thiserror::Error, Debug)] pub enum InitError { #[error("FFmpeg: {0}")] @@ -81,12 +94,19 @@ impl FragmentedAudioFile { Ok(Self { encoder, - output, + output: MuxSink::Direct(AudioOutput(output)), finished: false, has_frames: false, }) } + /// Hands encoded packets to a dedicated writer thread so a stalled + /// output volume no longer blocks the caller (an async runtime worker in + /// the recording pipeline). Bytes on disk and their order are unchanged. + pub fn enable_write_behind(&mut self) -> std::io::Result<()> { + self.output.enable_write_behind("audio", AUDIO_BUDGET_BYTES) + } + pub fn encoder(&self) -> &AACEncoder { &self.encoder } @@ -96,8 +116,16 @@ impl FragmentedAudioFile { frame: frame::Audio, timestamp: Duration, ) -> Result<(), ffmpeg::Error> { + if self.finished && self.output.is_write_behind() { + return Err(ffmpeg::Error::Exit); + } self.has_frames = true; - self.encoder.send_frame(frame, timestamp, &mut self.output) + self.encoder + .send_frame_to(frame, timestamp, &mut self.output) + } + + pub fn is_write_behind(&self) -> bool { + self.output.is_write_behind() } pub fn finish(&mut self) -> Result, FinishError> { @@ -114,12 +142,19 @@ impl FragmentedAudioFile { self.finished = true; - let flush_result = if self.has_frames { - self.encoder.flush(&mut self.output) - } else { - Ok(()) - }; - let trailer_result = self.output.write_trailer(); + let has_frames = self.has_frames; + let encoder = &mut self.encoder; + let flush_result = self + .output + .finish_flush(|output| { + if has_frames { + encoder.flush_to(output) + } else { + Ok(()) + } + }) + .map_err(|stalled| FinishError::WriteTrailerFailed(stalled.into()))?; + let trailer_result = self.output.with_target(|output| output.0.write_trailer()); finish_result(flush_result, trailer_result) } } @@ -220,6 +255,48 @@ mod publication_tests { output.finish().unwrap().unwrap(); } + #[test] + fn write_behind_audio_matches_inline_bytes() { + let directory = tempfile::tempdir().unwrap(); + let info = AudioInfo::new_raw(Sample::F32(Type::Packed), 48_000, 1); + let inline_path = directory.path().join("inline.m4a"); + let mut inline = FragmentedAudioFile::init(inline_path.clone(), info).unwrap(); + queue_tone(&mut inline, 300); + inline.finish().unwrap().unwrap(); + + let behind_path = directory.path().join("behind.m4a"); + let mut behind = FragmentedAudioFile::init(behind_path.clone(), info).unwrap(); + behind.enable_write_behind().unwrap(); + queue_tone(&mut behind, 300); + behind.output.drain(); + let mut input = format::input(&behind_path).unwrap(); + assert!(input.packets().next().is_some()); + drop(input); + behind.finish().unwrap().unwrap(); + + assert_eq!( + std::fs::read(&inline_path).unwrap(), + std::fs::read(&behind_path).unwrap() + ); + } + + #[test] + fn write_behind_audio_rejects_frames_after_finish() { + let directory = tempfile::tempdir().unwrap(); + let info = AudioInfo::new_raw(Sample::F32(Type::Packed), 48_000, 1); + let mut output = + FragmentedAudioFile::init(directory.path().join("finished.m4a"), info).unwrap(); + output.enable_write_behind().unwrap(); + queue_tone(&mut output, 10); + output.finish().unwrap().unwrap(); + let frame = + ffmpeg::frame::Audio::new(Sample::F32(Type::Packed), 1024, ffmpeg::ChannelLayout::MONO); + assert_eq!( + output.queue_frame(frame, Duration::from_secs(1)), + Err(ffmpeg::Error::Exit) + ); + } + #[test] fn short_fragmented_audio_remains_readable_after_finish() { let directory = tempfile::tempdir().unwrap(); @@ -284,7 +361,7 @@ mod publication_tests { fn set_output_error(output: &mut FragmentedAudioFile) { unsafe { - let context = output.output.as_mut_ptr(); + let context = output.output.with_target(|output| output.0.as_mut_ptr()); assert!(!(*context).pb.is_null()); (*(*context).pb).error = ffmpeg::Error::Other { errno: ffmpeg::error::EIO, diff --git a/crates/enc-ffmpeg/src/mux/mod.rs b/crates/enc-ffmpeg/src/mux/mod.rs index 24e53ac50fd..c0eb94e3aba 100644 --- a/crates/enc-ffmpeg/src/mux/mod.rs +++ b/crates/enc-ffmpeg/src/mux/mod.rs @@ -9,3 +9,4 @@ pub mod mp4; pub mod ogg; pub mod segmented_audio; pub mod segmented_stream; +pub(crate) mod write_behind; diff --git a/crates/enc-ffmpeg/src/mux/ogg.rs b/crates/enc-ffmpeg/src/mux/ogg.rs index c8d700c661f..e7a532777d7 100644 --- a/crates/enc-ffmpeg/src/mux/ogg.rs +++ b/crates/enc-ffmpeg/src/mux/ogg.rs @@ -1,14 +1,27 @@ use ffmpeg::{format, frame}; use std::{path::PathBuf, time::Duration}; +use super::write_behind::{AUDIO_BUDGET_BYTES, MuxSink, MuxTarget}; use crate::audio::opus::{OpusEncoder, OpusEncoderError}; pub struct OggFile { encoder: OpusEncoder, - output: format::context::Output, + output: MuxSink, finished: bool, } +struct OggOutput(format::context::Output); + +impl MuxTarget for OggOutput { + fn output(&self) -> &format::context::Output { + &self.0 + } + + fn output_mut(&mut self) -> &mut format::context::Output { + &mut self.0 + } +} + #[derive(thiserror::Error, Debug)] pub enum FinishError { #[error("Already finished")] @@ -32,11 +45,18 @@ impl OggFile { Ok(Self { encoder, - output, + output: MuxSink::Direct(OggOutput(output)), finished: false, }) } + /// Hands encoded packets to a dedicated writer thread so a stalled + /// output volume no longer blocks the caller. Bytes on disk and their + /// order are unchanged. + pub fn enable_write_behind(&mut self) -> std::io::Result<()> { + self.output.enable_write_behind("ogg", AUDIO_BUDGET_BYTES) + } + pub fn encoder(&self) -> &OpusEncoder { &self.encoder } @@ -46,7 +66,15 @@ impl OggFile { frame: frame::Audio, timestamp: Duration, ) -> Result<(), ffmpeg::Error> { - self.encoder.queue_frame(frame, timestamp, &mut self.output) + if self.finished && self.output.is_write_behind() { + return Err(ffmpeg::Error::Exit); + } + self.encoder + .queue_frame_to(frame, timestamp, &mut self.output) + } + + pub fn is_write_behind(&self) -> bool { + self.output.is_write_behind() } pub fn finish(&mut self) -> Result, FinishError> { @@ -56,9 +84,13 @@ impl OggFile { self.finished = true; - let flush_result = self.encoder.flush(&mut self.output); + let encoder = &mut self.encoder; + let flush_result = self + .output + .finish_flush(|output| encoder.flush_to(output)) + .map_err(|stalled| FinishError::WriteTrailerFailed(stalled.into()))?; self.output - .write_trailer() + .with_target(|output| output.0.write_trailer()) .map_err(FinishError::WriteTrailerFailed)?; Ok(flush_result) @@ -79,6 +111,15 @@ mod tests { use ffmpeg::ChannelLayout; fn write_recording(path: std::path::PathBuf, input_rate: u32, total_samples: usize) { + write_recording_with(path, input_rate, total_samples, false); + } + + fn write_recording_with( + path: std::path::PathBuf, + input_rate: u32, + total_samples: usize, + write_behind: bool, + ) { let info = AudioInfo::new_raw( format::Sample::F32(format::sample::Type::Packed), input_rate, @@ -86,6 +127,9 @@ mod tests { ); let mut file = OggFile::init(path, OpusEncoder::factory(info)).unwrap(); + if write_behind { + file.enable_write_behind().unwrap(); + } let chunk = 1024usize; let mut written = 0usize; @@ -142,6 +186,22 @@ mod tests { ); } + #[test] + fn write_behind_output_has_the_same_duration_and_size() { + let dir = tempfile::tempdir().unwrap(); + let inline = dir.path().join("inline.ogg"); + let behind = dir.path().join("behind.ogg"); + + write_recording_with(inline.clone(), 48_000, 48_000 * 5, false); + write_recording_with(behind.clone(), 48_000, 48_000 * 5, true); + + assert_eq!( + std::fs::metadata(&inline).unwrap().len(), + std::fs::metadata(&behind).unwrap().len() + ); + assert!((probed_duration_secs(&behind) - probed_duration_secs(&inline)).abs() < 1e-6); + } + #[test] fn sub_frame_input_is_flushed_not_dropped() { // Less input than one opus frame (20ms): finish() must drain the diff --git a/crates/enc-ffmpeg/src/mux/segmented_stream.rs b/crates/enc-ffmpeg/src/mux/segmented_stream.rs index 3b6091ceb5f..f32d6110ea7 100644 --- a/crates/enc-ffmpeg/src/mux/segmented_stream.rs +++ b/crates/enc-ffmpeg/src/mux/segmented_stream.rs @@ -1,4 +1,5 @@ use super::fragment_metadata::read_fragment_metadata; +use super::write_behind::{MuxSink, MuxTarget}; use cap_media_info::VideoInfo; use ffmpeg::{format, frame}; use serde::Serialize; @@ -76,10 +77,18 @@ pub struct SegmentedVideoEncoder { base_path: PathBuf, encoder: H264Encoder, + mux: MuxSink, + + segment_duration: Duration, + + disk_space_callback: Option, +} + +struct SegmentMux { + base_path: PathBuf, output: format::context::Output, current_index: u32, - segment_duration: Duration, completed_segments: Vec, @@ -87,9 +96,36 @@ pub struct SegmentedVideoEncoder { codec_info: CodecInfo, - disk_space_callback: Option, segment_tx: Option>, init_notified: bool, + initial_manifest_pending: bool, +} + +impl MuxTarget for SegmentMux { + fn output(&self) -> &format::context::Output { + &self.output + } + + fn output_mut(&mut self) -> &mut format::context::Output { + &mut self.output + } + + fn writer_started(&mut self) { + self.write_initial_manifest(); + } + + fn packet_written(&mut self) { + // The encoder holds one packet back to stamp real durations, so the + // init segment materializes a frame later than the first queue call; + // keep trying until it lands (no-op once notified). + self.try_notify_init_segment(); + + self.frames_since_segment_scan += 1; + if self.frames_since_segment_scan >= 10 { + self.frames_since_segment_scan = 0; + self.flush_pending_segments(); + } + } } #[derive(Debug, Clone)] @@ -197,6 +233,28 @@ impl SegmentedVideoEncoder { base_path: PathBuf, video_config: VideoInfo, config: SegmentedVideoEncoderConfig, + ) -> Result { + Self::init_with_output_mode(base_path, video_config, config, false) + } + + /// Like [`Self::init`], but muxing, segment publication and manifest + /// writes (including the initial manifest) run on a dedicated writer + /// thread with a memory budget sized from the stream bitrate, so storage + /// stalls stop blocking the encoder. Output bytes and their order are + /// unchanged. Falls back to inline muxing if the thread cannot start. + pub fn init_write_behind( + base_path: PathBuf, + video_config: VideoInfo, + config: SegmentedVideoEncoderConfig, + ) -> Result { + Self::init_with_output_mode(base_path, video_config, config, true) + } + + fn init_with_output_mode( + base_path: PathBuf, + video_config: VideoInfo, + config: SegmentedVideoEncoderConfig, + write_behind: bool, ) -> Result { std::fs::create_dir_all(&base_path)?; @@ -295,21 +353,38 @@ impl SegmentedVideoEncoder { "Initialized segmented video encoder with FFmpeg DASH muxer (init.mp4 + m4s segments)" ); - let instance = Self { - base_path, - encoder, + let mux = SegmentMux { + base_path: base_path.clone(), output, current_index: 1, - segment_duration: config.segment_duration, completed_segments: Vec::new(), frames_since_segment_scan: 0, codec_info, - disk_space_callback: None, segment_tx: None, init_notified: false, + initial_manifest_pending: true, }; - instance.write_in_progress_manifest(); + let mut instance = Self { + base_path, + encoder, + mux: MuxSink::Direct(mux), + segment_duration: config.segment_duration, + disk_space_callback: None, + }; + + if write_behind { + let budget = super::write_behind::video_budget_bytes(instance.encoder.bit_rate()); + match instance.mux.enable_write_behind("video", budget) { + Ok(()) => tracing::debug!(budget_bytes = budget, "Video write-behind enabled"), + Err(error) => { + tracing::warn!(%error, "Video write-behind unavailable, muxing inline"); + } + } + } + if !instance.mux.is_write_behind() { + instance.mux.with_target(SegmentMux::write_initial_manifest); + } Ok(instance) } @@ -319,28 +394,10 @@ impl SegmentedVideoEncoder { } pub fn set_segment_callback(&mut self, tx: std::sync::mpsc::Sender) { - self.segment_tx = Some(tx); - self.try_notify_init_segment(); - } - - fn try_notify_init_segment(&mut self) { - if self.init_notified { - return; - } - let init_path = self.init_segment_path(); - if let Ok(meta) = std::fs::metadata(&init_path) - && meta.len() > 0 - { - self.init_notified = true; - self.notify_segment(SegmentCompletedEvent { - path: init_path, - index: 0, - duration: 0.0, - file_size: meta.len(), - is_init: true, - media_type: SegmentMediaType::Video, - }); - } + self.mux.with_target(|mux| { + mux.segment_tx = Some(tx); + mux.try_notify_init_segment(); + }); } /// Whether frames should be queued as CVPixelBuffers via @@ -377,20 +434,114 @@ impl SegmentedVideoEncoder { // stream restarts, dropped frames) stay in the timeline instead of // compressing it and drifting video ahead of audio. self.encoder - .queue_frame(frame, timestamp, &mut self.output)?; + .queue_frame_to(frame, timestamp, &mut self.mux)?; - // The encoder holds one packet back to stamp real durations, so the - // init segment materializes a frame later than the first queue call; - // keep trying until it lands (no-op once notified). - self.try_notify_init_segment(); + self.mux.frame_queued(); - self.frames_since_segment_scan += 1; - if self.frames_since_segment_scan >= 10 { - self.frames_since_segment_scan = 0; - self.flush_pending_segments(); + Ok(()) + } + + pub fn finish(&mut self) -> Result<(), FinishError> { + let encoder = &mut self.encoder; + let flush_result = self + .mux + .finish_flush(|mux| encoder.flush_to(mux)) + .map_err(std::io::Error::from)? + .inspect_err(|error| { + tracing::warn!(%error, "Video encoder flush failed"); + }); + self.mux.with_target(|mux| mux.finish(flush_result)) + } + + pub fn finish_with_timestamp(&mut self, _timestamp: Duration) -> Result<(), FinishError> { + self.finish() + } + + pub fn completed_segments(&self) -> Vec { + self.mux + .with_target_ref(|mux| mux.completed_segments.clone()) + } + + pub fn current_encoder(&self) -> Option<&H264Encoder> { + Some(&self.encoder) + } + + pub fn current_encoder_mut(&mut self) -> Option<&mut H264Encoder> { + Some(&mut self.encoder) + } + + pub fn base_path(&self) -> &Path { + &self.base_path + } + + pub fn segment_duration(&self) -> Duration { + self.segment_duration + } + + pub fn current_index(&self) -> u32 { + self.mux.with_target_ref(|mux| mux.current_index) + } + + pub fn init_segment_path(&self) -> PathBuf { + self.base_path.join(INIT_SEGMENT_NAME) + } + + pub fn validate_init_segment(&self) -> Result<(), String> { + let init_path = self.init_segment_path(); + + if !init_path.exists() { + return Err(format!( + "init.mp4 is missing at {}. M4S segments will be unplayable without it.", + init_path.display() + )); } - Ok(()) + match std::fs::metadata(&init_path) { + Ok(metadata) => { + let size = metadata.len(); + if size < 100 { + return Err(format!( + "init.mp4 at {} is too small ({} bytes). It may be corrupted.", + init_path.display(), + size + )); + } + Ok(()) + } + Err(e) => Err(format!( + "Cannot read init.mp4 metadata at {}: {}", + init_path.display(), + e + )), + } + } +} + +impl SegmentMux { + fn write_initial_manifest(&mut self) { + if std::mem::take(&mut self.initial_manifest_pending) { + self.write_in_progress_manifest(); + } + } + + fn try_notify_init_segment(&mut self) { + if self.init_notified { + return; + } + let init_path = self.base_path.join(INIT_SEGMENT_NAME); + if let Ok(meta) = std::fs::metadata(&init_path) + && meta.len() > 0 + { + self.init_notified = true; + self.notify_segment(SegmentCompletedEvent { + path: init_path, + index: 0, + duration: 0.0, + file_size: meta.len(), + is_init: true, + media_type: SegmentMediaType::Video, + }); + } } fn notify_segment(&self, event: SegmentCompletedEvent) { @@ -493,10 +644,7 @@ impl SegmentedVideoEncoder { } } - pub fn finish(&mut self) -> Result<(), FinishError> { - let flush_result = self.encoder.flush(&mut self.output).inspect_err(|error| { - tracing::warn!(%error, "Video encoder flush failed"); - }); + fn finish(&mut self, flush_result: Result<(), ffmpeg::Error>) -> Result<(), FinishError> { let trailer_result = self.output.write_trailer().inspect_err(|error| { tracing::warn!(%error, "Video trailer publication failed"); }); @@ -520,10 +668,6 @@ impl SegmentedVideoEncoder { Ok(()) } - pub fn finish_with_timestamp(&mut self, _timestamp: Duration) -> Result<(), FinishError> { - self.finish() - } - fn finalize_pending_tmp_files(&self) { let Ok(entries) = std::fs::read_dir(&self.base_path) else { return; @@ -688,64 +832,6 @@ impl SegmentedVideoEncoder { let manifest_path = self.base_path.join("manifest.json"); atomic_write_json(&manifest_path, &manifest) } - - pub fn completed_segments(&self) -> &[VideoSegmentInfo] { - &self.completed_segments - } - - pub fn current_encoder(&self) -> Option<&H264Encoder> { - Some(&self.encoder) - } - - pub fn current_encoder_mut(&mut self) -> Option<&mut H264Encoder> { - Some(&mut self.encoder) - } - - pub fn base_path(&self) -> &Path { - &self.base_path - } - - pub fn segment_duration(&self) -> Duration { - self.segment_duration - } - - pub fn current_index(&self) -> u32 { - self.current_index - } - - pub fn init_segment_path(&self) -> PathBuf { - self.base_path.join(INIT_SEGMENT_NAME) - } - - pub fn validate_init_segment(&self) -> Result<(), String> { - let init_path = self.init_segment_path(); - - if !init_path.exists() { - return Err(format!( - "init.mp4 is missing at {}. M4S segments will be unplayable without it.", - init_path.display() - )); - } - - match std::fs::metadata(&init_path) { - Ok(metadata) => { - let size = metadata.len(); - if size < 100 { - return Err(format!( - "init.mp4 at {} is too small ({} bytes). It may be corrupted.", - init_path.display(), - size - )); - } - Ok(()) - } - Err(e) => Err(format!( - "Cannot read init.mp4 metadata at {}: {}", - init_path.display(), - e - )), - } - } } #[cfg(test)] @@ -838,6 +924,100 @@ mod tests { ); } + fn record_segments(base_path: &Path, write_behind: bool) -> Vec<(u32, bool, u64)> { + let init = if write_behind { + SegmentedVideoEncoder::init_write_behind + } else { + SegmentedVideoEncoder::init + }; + let mut encoder = init( + base_path.to_path_buf(), + test_video_info(), + SegmentedVideoEncoderConfig::default(), + ) + .unwrap(); + assert_eq!(encoder.mux.is_write_behind(), write_behind); + let (tx, rx) = mpsc::channel(); + encoder.set_segment_callback(tx); + for i in 0..200 { + encoder + .queue_frame( + create_test_frame(320, 240), + Duration::from_micros(i * 33_320), + ) + .unwrap(); + } + encoder + .finish_with_timestamp(Duration::from_secs(7)) + .unwrap(); + drop(encoder); + rx.try_iter() + .map(|event| { + assert_eq!( + event.file_size, + std::fs::metadata(&event.path).unwrap().len() + ); + (event.index, event.is_init, (event.duration * 1e6) as u64) + }) + .collect() + } + + fn published_index(base_path: &Path) -> (Vec, String, serde_json::Value) { + let mut names: Vec = std::fs::read_dir(base_path) + .unwrap() + .flatten() + .map(|entry| entry.file_name().to_string_lossy().into_owned()) + .collect(); + names.sort(); + let mpd = std::fs::read_to_string(base_path.join("dash_manifest.mpd")).unwrap(); + let mut manifest: serde_json::Value = + serde_json::from_slice(&std::fs::read(base_path.join("manifest.json")).unwrap()) + .unwrap(); + for segment in manifest["segments"].as_array_mut().unwrap() { + segment.as_object_mut().unwrap().remove("file_size"); + } + (names, mpd, manifest) + } + + #[test] + fn write_behind_publishes_the_initial_manifest_from_the_writer_thread() { + ffmpeg::init().unwrap(); + let temp = tempfile::tempdir().unwrap(); + let encoder = SegmentedVideoEncoder::init_write_behind( + temp.path().to_path_buf(), + test_video_info(), + SegmentedVideoEncoderConfig::default(), + ) + .unwrap(); + let manifest_path = temp.path().join("manifest.json"); + let deadline = std::time::Instant::now() + Duration::from_secs(2); + while !manifest_path.exists() && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(5)); + } + let manifest: serde_json::Value = + serde_json::from_slice(&std::fs::read(&manifest_path).unwrap()).unwrap(); + assert_eq!(manifest["is_complete"], false); + assert_eq!(manifest["segments"][0]["index"], 1); + drop(encoder); + } + + #[test] + fn write_behind_publishes_the_same_segments_as_inline_muxing() { + ffmpeg::init().unwrap(); + let temp = tempfile::tempdir().unwrap(); + let inline = temp.path().join("inline"); + let behind = temp.path().join("behind"); + + let inline_events = record_segments(&inline, false); + let behind_events = record_segments(&behind, true); + + // VideoToolbox payload sizes are not reproducible across sessions, so + // the published timeline is compared rather than media bytes. + assert!(inline_events.len() >= 3); + assert_eq!(inline_events, behind_events); + assert_eq!(published_index(&inline), published_index(&behind)); + } + #[test] fn failed_dash_manifest_publication_retains_video_without_claiming_completion() { ffmpeg::init().unwrap(); diff --git a/crates/enc-ffmpeg/src/mux/write_behind.rs b/crates/enc-ffmpeg/src/mux/write_behind.rs new file mode 100644 index 00000000000..d2260cf275b --- /dev/null +++ b/crates/enc-ffmpeg/src/mux/write_behind.rs @@ -0,0 +1,892 @@ +//! Write-behind muxing: encoders hand finished packets to a dedicated writer +//! thread that owns the `AVFormatContext`, so a stalled output volume (slow +//! SD card, external or network drive) blocks only that thread instead of +//! the encoder or the async runtime driving it. +//! +//! Packets are written in exactly the order they were produced, so the +//! on-disk bytes are identical to writing inline. Memory is bounded by a byte +//! budget; once it is full the producer blocks until the writer catches up, +//! which is the inline behaviour. +//! +//! Finishing drains the queue while storage keeps making progress, bounded +//! by [`DRAIN_NO_PROGRESS_LIMIT`] without a completed write and +//! [`DRAIN_TOTAL_LIMIT`] overall. Past the total limit the unwritten tail is +//! discarded and the output is finalized normally ([`DrainOutcome::Truncated`]). +//! When no write completes for the no-progress limit, finishing returns +//! [`DrainOutcome::Stuck`] without the trailer or final manifest, leaving the +//! files in the state recovery expects from an interrupted recording. As with +//! an encoder blocked in an inline write, the queued packets still reach disk +//! once storage recovers. Dropping a writer that was never finished discards +//! its queue instead and touches nothing after its current call returns. + +use crate::base::PacketSink; +use ffmpeg::{Packet, Rational, format}; +use std::{ + collections::VecDeque, + sync::{ + Arc, Condvar, Mutex, MutexGuard, + atomic::{AtomicBool, Ordering}, + }, + thread::JoinHandle, + time::{Duration, Instant}, +}; + +const VIDEO_BUFFER_WINDOW_SECS: u64 = 40; +const MIN_VIDEO_BUDGET_BYTES: usize = 32 * 1024 * 1024; +const MAX_VIDEO_BUDGET_BYTES: usize = 512 * 1024 * 1024; +const VIDEO_BUDGET_MEMORY_DIVISOR: u64 = 32; +pub(crate) const AUDIO_BUDGET_BYTES: usize = 16 * 1024 * 1024; +const STALL_LOG_THRESHOLD: Duration = Duration::from_secs(1); +const WRITE_ERROR_LOG_INTERVAL: u64 = 100; +// Muxers join their encoder thread for up to 5 s before finishing, and an +// unjoined pipeline stop gives up after 10 s; draining for at most 4 s keeps +// the trailer and manifest inside that window. +pub(crate) const DRAIN_NO_PROGRESS_LIMIT: Duration = Duration::from_secs(4); +pub(crate) const DRAIN_TOTAL_LIMIT: Duration = Duration::from_secs(4); +const DROP_JOIN_LIMIT: Duration = Duration::from_millis(250); +const WAIT_POLL: Duration = Duration::from_millis(50); + +pub(crate) fn video_budget_bytes(bits_per_second: u64) -> usize { + video_budget_for(bits_per_second, total_memory_bytes()) +} + +fn video_budget_for(bits_per_second: u64, total_memory: Option) -> usize { + let ceiling = total_memory + .map(|memory| usize::try_from(memory / VIDEO_BUDGET_MEMORY_DIVISOR).unwrap_or(usize::MAX)) + .unwrap_or(MAX_VIDEO_BUDGET_BYTES) + .clamp(MIN_VIDEO_BUDGET_BYTES, MAX_VIDEO_BUDGET_BYTES); + let bytes = (bits_per_second / 8).saturating_mul(VIDEO_BUFFER_WINDOW_SECS); + usize::try_from(bytes) + .unwrap_or(usize::MAX) + .clamp(MIN_VIDEO_BUDGET_BYTES, ceiling) +} + +fn total_memory_bytes() -> Option { + let system = sysinfo::System::new_with_specifics( + sysinfo::RefreshKind::new().with_memory(sysinfo::MemoryRefreshKind::new().with_ram()), + ); + Some(system.total_memory()).filter(|memory| *memory > 0) +} + +pub(crate) trait MuxTarget: Send + 'static { + fn output(&self) -> &format::context::Output; + + fn output_mut(&mut self) -> &mut format::context::Output; + + fn writer_started(&mut self) {} + + fn packet_written(&mut self) {} +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct StorageStalled; + +impl From for std::io::Error { + fn from(_: StorageStalled) -> Self { + std::io::Error::new( + std::io::ErrorKind::TimedOut, + "output storage stalled while finishing; recording left for recovery", + ) + } +} + +impl From for ffmpeg::Error { + fn from(_: StorageStalled) -> Self { + ffmpeg::Error::Other { + errno: ffmpeg::error::EIO, + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DrainOutcome { + Complete, + Truncated { packets: usize, bytes: usize }, + Stuck, +} + +pub(crate) enum MuxSink { + Direct(M), + Behind(WriteBehind), + Detached, +} + +impl MuxSink { + pub fn enable_write_behind( + &mut self, + label: &'static str, + budget: usize, + ) -> std::io::Result<()> { + if !matches!(self, Self::Direct(_)) { + return Ok(()); + } + let Self::Direct(target) = std::mem::replace(self, Self::Detached) else { + unreachable!("checked above"); + }; + match WriteBehind::spawn(label, target, budget, DrainLimits::DEFAULT) { + Ok(writer) => { + *self = Self::Behind(writer); + Ok(()) + } + Err((target, error)) => { + *self = Self::Direct(target); + Err(error) + } + } + } + + pub fn is_write_behind(&self) -> bool { + matches!(self, Self::Behind(_)) + } + + pub fn with_target(&mut self, f: impl FnOnce(&mut M) -> R) -> R { + match self { + Self::Direct(target) => f(target), + Self::Behind(writer) => f(&mut writer.target()), + Self::Detached => unreachable!("mux sink detached"), + } + } + + pub fn with_target_ref(&self, f: impl FnOnce(&M) -> R) -> R { + match self { + Self::Direct(target) => f(target), + Self::Behind(writer) => f(&writer.target()), + Self::Detached => unreachable!("mux sink detached"), + } + } + + pub fn frame_queued(&mut self) { + if let Self::Direct(target) = self { + target.packet_written(); + } + } + + /// Marks the following packets as the end-of-stream flush: they bypass + /// the budget and their write errors are reported by [`Self::close`], + /// as an inline flush would report them. + pub fn begin_finish(&mut self) { + if let Self::Behind(writer) = self { + writer.begin_finish(); + } + } + + /// Drains the queued packets within the drain limits. An `Err` is the + /// first write error among the end-of-stream packets. + pub fn close(&mut self) -> Result { + match self { + Self::Behind(writer) => writer.close(), + Self::Direct(_) | Self::Detached => Ok(DrainOutcome::Complete), + } + } + + /// Runs the encoder's end-of-stream `flush` into this sink and drains + /// the writer. `Err` means storage stalled: the output was abandoned and + /// must not be finalized. + pub fn finish_flush( + &mut self, + flush: impl FnOnce(&mut Self) -> Result<(), ffmpeg::Error>, + ) -> Result, StorageStalled> { + self.begin_finish(); + let flush_result = flush(self); + match self.close() { + Ok(DrainOutcome::Stuck) => Err(StorageStalled), + Ok(DrainOutcome::Complete | DrainOutcome::Truncated { .. }) => Ok(flush_result), + Err(error) => Ok(flush_result.and(Err(error))), + } + } + + #[cfg(test)] + pub fn drain(&self) { + if let Self::Behind(writer) = self { + writer.drain(); + } + } +} + +impl PacketSink for MuxSink { + fn stream_time_base(&self, stream_index: usize) -> Rational { + match self { + Self::Direct(target) => target.output().stream_time_base(stream_index), + Self::Behind(writer) => writer.time_bases[stream_index], + Self::Detached => unreachable!("mux sink detached"), + } + } + + fn write_packet(&mut self, packet: Packet) -> Result<(), ffmpeg::Error> { + match self { + Self::Direct(target) => packet.write_interleaved(target.output_mut()), + Self::Behind(writer) => writer.write_packet(packet), + Self::Detached => unreachable!("mux sink detached"), + } + } +} + +#[derive(Clone, Copy)] +struct DrainLimits { + no_progress: Duration, + total: Duration, +} + +impl DrainLimits { + const DEFAULT: Self = Self { + no_progress: DRAIN_NO_PROGRESS_LIMIT, + total: DRAIN_TOTAL_LIMIT, + }; +} + +struct Queue { + packets: VecDeque<(Packet, bool)>, + bytes: usize, + in_flight: bool, + closed: bool, + finishing: bool, + exited: bool, + completed_writes: u64, + write_errors: u64, + final_error: Option, +} + +struct Shared { + queue: Mutex, + changed: Condvar, + abandoned: AtomicBool, + target: Mutex, +} + +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +impl Shared { + fn wait<'a>(&self, queue: MutexGuard<'a, Queue>, timeout: Duration) -> MutexGuard<'a, Queue> { + self.changed + .wait_timeout(queue, timeout) + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .0 + } + + fn abandon(&self, queue: &mut Queue) -> (usize, usize) { + self.abandoned.store(true, Ordering::Release); + let packets = queue.packets.len(); + let bytes = queue + .packets + .drain(..) + .map(|(packet, _)| packet.size()) + .sum::(); + queue.bytes -= bytes; + self.changed.notify_all(); + (packets, bytes) + } +} + +pub(crate) struct WriteBehind { + shared: Arc>, + time_bases: Vec, + budget: usize, + label: &'static str, + limits: DrainLimits, + thread: Option>, +} + +impl WriteBehind { + fn spawn( + label: &'static str, + target: M, + budget: usize, + limits: DrainLimits, + ) -> Result { + let time_bases = target + .output() + .streams() + .map(|stream| stream.time_base()) + .collect(); + let shared = Arc::new(Shared { + queue: Mutex::new(Queue { + packets: VecDeque::new(), + bytes: 0, + in_flight: false, + closed: false, + finishing: false, + exited: false, + completed_writes: 0, + write_errors: 0, + final_error: None, + }), + changed: Condvar::new(), + abandoned: AtomicBool::new(false), + target: Mutex::new(target), + }); + let spawned = std::thread::Builder::new() + .name(format!("mux-writer-{label}")) + .spawn({ + let shared = shared.clone(); + move || run_writer(&shared, label) + }); + let thread = match spawned { + Ok(thread) => thread, + Err(error) => { + let target = Arc::try_unwrap(shared) + .ok() + .expect("writer thread never started") + .target + .into_inner() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + return Err((target, error)); + } + }; + Ok(Self { + shared, + time_bases, + budget, + label, + limits, + thread: Some(thread), + }) + } + + fn target(&self) -> MutexGuard<'_, M> { + lock(&self.shared.target) + } + + fn begin_finish(&mut self) { + lock(&self.shared.queue).finishing = true; + } + + fn write_packet(&mut self, packet: Packet) -> Result<(), ffmpeg::Error> { + let size = packet.size(); + let mut queue = lock(&self.shared.queue); + let mut blocked_since = None; + loop { + if queue.exited { + return Err(ffmpeg::Error::Bug); + } + if queue.finishing || queue.bytes == 0 || queue.bytes + size <= self.budget { + break; + } + if blocked_since.is_none() { + blocked_since = Some(Instant::now()); + tracing::warn!( + writer = self.label, + queued_bytes = queue.bytes, + budget_bytes = self.budget, + "Output write-behind budget exhausted; encoder now waits for storage" + ); + } + queue = self.shared.wait(queue, WAIT_POLL); + } + queue.bytes += size; + let final_packet = queue.finishing; + queue.packets.push_back((packet, final_packet)); + drop(queue); + self.shared.changed.notify_all(); + + if let Some(since) = blocked_since { + tracing::warn!( + writer = self.label, + waited_ms = since.elapsed().as_millis() as u64, + "Output write-behind budget recovered" + ); + } + + Ok(()) + } + + #[cfg(test)] + fn drain(&self) { + let mut queue = lock(&self.shared.queue); + while (!queue.packets.is_empty() || queue.in_flight) && !queue.exited { + queue = self.shared.wait(queue, WAIT_POLL); + } + } + + fn close(&mut self) -> Result { + let Some(thread) = self.thread.take() else { + return Ok(DrainOutcome::Complete); + }; + let started = Instant::now(); + let mut last_progress = started; + let mut queue = lock(&self.shared.queue); + queue.closed = true; + self.shared.changed.notify_all(); + let mut seen_writes = queue.completed_writes; + let mut truncated = None; + while !queue.exited { + if queue.completed_writes != seen_writes { + seen_writes = queue.completed_writes; + last_progress = Instant::now(); + } + if last_progress.elapsed() >= self.limits.no_progress { + tracing::error!( + writer = self.label, + stalled_ms = last_progress.elapsed().as_millis() as u64, + queued_packets = queue.packets.len(), + queued_bytes = queue.bytes, + "Output storage made no progress while finishing; leaving the recording for recovery and writing the buffer once storage recovers" + ); + return Ok(DrainOutcome::Stuck); + } + if truncated.is_none() && started.elapsed() >= self.limits.total { + let (packets, bytes) = self.shared.abandon(&mut queue); + tracing::warn!( + writer = self.label, + discarded_packets = packets, + discarded_bytes = bytes, + "Output storage too slow to drain the write-behind buffer; finishing without its tail" + ); + truncated = Some(DrainOutcome::Truncated { packets, bytes }); + } + queue = self.shared.wait(queue, WAIT_POLL); + } + let final_error = queue.final_error.take(); + drop(queue); + let panicked = thread.join().is_err(); + match final_error { + Some(error) => Err(error), + None if panicked => Err(ffmpeg::Error::Bug), + None => Ok(truncated.unwrap_or(DrainOutcome::Complete)), + } + } +} + +impl Drop for WriteBehind { + fn drop(&mut self) { + let Some(thread) = self.thread.take() else { + return; + }; + let mut queue = lock(&self.shared.queue); + queue.closed = true; + self.shared.abandon(&mut queue); + let deadline = Instant::now() + DROP_JOIN_LIMIT; + while !queue.exited && Instant::now() < deadline { + queue = self.shared.wait(queue, WAIT_POLL.min(DROP_JOIN_LIMIT)); + } + let exited = queue.exited; + drop(queue); + if exited { + let _ = thread.join(); + } + } +} + +struct WriterExit<'a, M>(&'a Shared); + +impl Drop for WriterExit<'_, M> { + fn drop(&mut self) { + let mut queue = lock(&self.0.queue); + queue.exited = true; + queue.in_flight = false; + drop(queue); + self.0.changed.notify_all(); + } +} + +fn run_writer(shared: &Shared, label: &'static str) { + let _exit = WriterExit(shared); + { + let mut target = lock(&shared.target); + if !shared.abandoned.load(Ordering::Acquire) { + target.writer_started(); + } + } + loop { + let next = { + let mut queue = lock(&shared.queue); + loop { + if shared.abandoned.load(Ordering::Acquire) { + break None; + } + if let Some(next) = queue.packets.pop_front() { + queue.in_flight = true; + break Some(next); + } + if queue.closed { + break None; + } + queue = shared + .changed + .wait(queue) + .unwrap_or_else(|poisoned| poisoned.into_inner()); + } + }; + let Some((packet, final_packet)) = next else { + return; + }; + + let size = packet.size(); + let started = Instant::now(); + let result = { + let mut target = lock(&shared.target); + if shared.abandoned.load(Ordering::Acquire) { + None + } else { + let result = packet.write_interleaved(target.output_mut()); + if !shared.abandoned.load(Ordering::Acquire) { + target.packet_written(); + } + Some(result) + } + }; + let elapsed = started.elapsed(); + + let (backlog, error_count) = { + let mut queue = lock(&shared.queue); + queue.bytes -= size; + queue.in_flight = false; + queue.completed_writes += 1; + if let Some(Err(error)) = &result { + queue.write_errors += 1; + if final_packet && queue.final_error.is_none() { + queue.final_error = Some(*error); + } + } + (queue.bytes, queue.write_errors) + }; + shared.changed.notify_all(); + + if let Some(Err(error)) = result + && (error_count == 1 || error_count.is_multiple_of(WRITE_ERROR_LOG_INTERVAL)) + { + tracing::warn!( + writer = label, + %error, + failed_writes = error_count, + "Failed to write output packet" + ); + } + + if elapsed >= STALL_LOG_THRESHOLD { + tracing::warn!( + writer = label, + stalled_ms = elapsed.as_millis() as u64, + backlog_bytes = backlog, + "Output storage write stalled; capture continued into the write-behind buffer" + ); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + struct NutTarget { + output: format::context::Output, + write_delay: Duration, + panic_on_write: bool, + } + + impl MuxTarget for NutTarget { + fn output(&self) -> &format::context::Output { + &self.output + } + + fn output_mut(&mut self) -> &mut format::context::Output { + assert!(!self.panic_on_write, "simulated writer failure"); + std::thread::sleep(self.write_delay); + &mut self.output + } + } + + fn nut_target(path: &std::path::Path) -> NutTarget { + ffmpeg::init().unwrap(); + let mut output = format::output_as(path, "nut").unwrap(); + let codec = ffmpeg::encoder::find(ffmpeg::codec::Id::PCM_S16LE).unwrap(); + let mut stream = output.add_stream(codec).unwrap(); + stream.set_time_base(Rational::new(1, 48_000)); + unsafe { + let parameters = (*stream.as_mut_ptr()).codecpar; + (*parameters).codec_type = ffmpeg::ffi::AVMediaType::AVMEDIA_TYPE_AUDIO; + (*parameters).codec_id = ffmpeg::ffi::AVCodecID::AV_CODEC_ID_PCM_S16LE; + (*parameters).sample_rate = 48_000; + ffmpeg::ffi::av_channel_layout_default(&mut (*parameters).ch_layout, 1); + } + output.write_header().unwrap(); + NutTarget { + output, + write_delay: Duration::ZERO, + panic_on_write: false, + } + } + + fn packet(pts: i64, size: usize) -> Packet { + let mut packet = Packet::copy(&vec![0u8; size]); + packet.set_stream(0); + packet.set_pts(Some(pts)); + packet.set_dts(Some(pts)); + packet.set_duration(1024); + packet + } + + fn invalid_packet(pts: i64) -> Packet { + let mut packet = packet(pts, 16); + packet.set_stream(7); + packet + } + + fn behind(target: NutTarget, budget: usize, limits: DrainLimits) -> MuxSink { + MuxSink::Behind( + WriteBehind::spawn("test", target, budget, limits) + .ok() + .unwrap(), + ) + } + + fn shared(sink: &MuxSink) -> Arc> { + match sink { + MuxSink::Behind(writer) => writer.shared.clone(), + _ => panic!("write-behind not enabled"), + } + } + + fn header_only_len(directory: &std::path::Path) -> u64 { + let path = directory.join("header-only.nut"); + drop(nut_target(&path)); + std::fs::metadata(path).unwrap().len() + } + + fn written_pts(path: &std::path::Path) -> Vec { + let mut input = format::input(path).unwrap(); + input + .packets() + .filter_map(|(_, packet)| packet.pts()) + .collect() + } + + #[test] + fn producer_runs_ahead_of_a_stalled_writer_and_order_is_preserved() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("out.nut"); + let mut sink = MuxSink::Direct(nut_target(&path)); + sink.enable_write_behind("test", 1 << 20).unwrap(); + assert!(sink.is_write_behind()); + + let shared = shared(&sink); + let stalled = lock(&shared.target); + let started = Instant::now(); + for pts in 0..64 { + sink.write_packet(packet(pts * 1024, 2048)).unwrap(); + } + assert!(started.elapsed() < Duration::from_secs(1)); + drop(stalled); + + assert_eq!(sink.close(), Ok(DrainOutcome::Complete)); + sink.with_target(|target| target.output.write_trailer().unwrap()); + assert_eq!( + written_pts(&path), + (0..64).map(|pts| pts * 1024).collect::>() + ); + } + + #[test] + fn exhausted_budget_falls_back_to_waiting_for_storage() { + let directory = tempfile::tempdir().unwrap(); + let mut sink = MuxSink::Direct(nut_target(&directory.path().join("out.nut"))); + sink.enable_write_behind("test", 8 * 1024).unwrap(); + let shared = shared(&sink); + + let stalled = lock(&shared.target); + for pts in 0..4 { + sink.write_packet(packet(pts * 1024, 2048)).unwrap(); + } + let producer = std::thread::spawn(move || { + let started = Instant::now(); + sink.write_packet(packet(4 * 1024, 2048)).unwrap(); + (sink, started.elapsed()) + }); + std::thread::sleep(Duration::from_millis(300)); + drop(stalled); + let (mut sink, waited) = producer.join().unwrap(); + assert!(waited >= Duration::from_millis(250)); + assert_eq!(sink.close(), Ok(DrainOutcome::Complete)); + } + + #[test] + fn oversized_packet_is_accepted_into_an_empty_queue() { + let directory = tempfile::tempdir().unwrap(); + let mut sink = MuxSink::Direct(nut_target(&directory.path().join("out.nut"))); + sink.enable_write_behind("test", 1024).unwrap(); + sink.write_packet(packet(0, 4096)).unwrap(); + assert_eq!(sink.close(), Ok(DrainOutcome::Complete)); + } + + #[test] + fn earlier_write_errors_never_abort_the_end_of_stream_flush() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("out.nut"); + let mut sink = MuxSink::Direct(nut_target(&path)); + sink.enable_write_behind("test", 1 << 20).unwrap(); + + sink.write_packet(packet(0, 64)).unwrap(); + sink.write_packet(invalid_packet(1024)).unwrap(); + sink.drain(); + sink.write_packet(packet(2048, 64)).unwrap(); + sink.begin_finish(); + sink.write_packet(packet(3072, 64)).unwrap(); + + assert_eq!(sink.close(), Ok(DrainOutcome::Complete)); + sink.with_target(|target| target.output.write_trailer().unwrap()); + assert_eq!(written_pts(&path), vec![0, 2048, 3072]); + } + + #[test] + fn end_of_stream_write_errors_are_reported_like_an_inline_flush() { + let directory = tempfile::tempdir().unwrap(); + let mut sink = MuxSink::Direct(nut_target(&directory.path().join("out.nut"))); + sink.enable_write_behind("test", 1 << 20).unwrap(); + sink.write_packet(packet(0, 64)).unwrap(); + sink.begin_finish(); + sink.write_packet(invalid_packet(1024)).unwrap(); + assert!(sink.close().is_err()); + } + + #[test] + fn finish_returns_on_stuck_storage_and_keeps_the_buffer_for_when_it_recovers() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("out.nut"); + let packets = 8usize; + let packet_size = 64 * 1024; + { + let target = nut_target(&path); + let limits = DrainLimits { + no_progress: Duration::from_millis(200), + total: Duration::from_secs(30), + }; + let mut sink = behind(target, 1 << 20, limits); + let shared = shared(&sink); + let stalled = lock(&shared.target); + for pts in 0..packets { + sink.write_packet(packet(pts as i64 * 1024, packet_size)) + .unwrap(); + } + let started = Instant::now(); + assert_eq!(sink.close(), Ok(DrainOutcome::Stuck)); + assert!(started.elapsed() < Duration::from_secs(2)); + drop(sink); + drop(stalled); + let deadline = Instant::now() + Duration::from_secs(2); + while Arc::strong_count(&shared) > 1 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(Arc::strong_count(&shared), 1); + } + assert!( + std::fs::metadata(&path).unwrap().len() + >= header_only_len(directory.path()) + (packets * packet_size) as u64 + ); + } + + #[test] + fn slow_storage_drains_within_the_total_limit_then_finishes_without_the_tail() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("out.nut"); + let mut target = nut_target(&path); + target.write_delay = Duration::from_millis(60); + let limits = DrainLimits { + no_progress: Duration::from_secs(2), + total: Duration::from_millis(300), + }; + let mut sink = behind(target, 1 << 20, limits); + for pts in 0..40 { + sink.write_packet(packet(pts * 1024, 64)).unwrap(); + } + let started = Instant::now(); + let outcome = sink.close().unwrap(); + assert!(started.elapsed() < Duration::from_secs(2)); + let DrainOutcome::Truncated { packets, .. } = outcome else { + panic!("expected a truncated drain, got {outcome:?}"); + }; + assert!(packets > 0); + sink.with_target(|target| { + target.write_delay = Duration::ZERO; + target.output.write_trailer().unwrap(); + }); + let pts = written_pts(&path); + assert_eq!(pts.len() + packets, 40); + assert_eq!( + pts, + (0..pts.len() as i64) + .map(|pts| pts * 1024) + .collect::>() + ); + } + + #[test] + fn dropping_without_finish_stops_the_writer_promptly() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("out.nut"); + let shared; + { + let mut sink = MuxSink::Direct(nut_target(&path)); + sink.enable_write_behind("test", 1 << 20).unwrap(); + shared = self::shared(&sink); + let stalled = lock(&shared.target); + for pts in 0..16 { + sink.write_packet(packet(pts * 1024, 64 * 1024)).unwrap(); + } + let dropping = std::thread::spawn(move || { + let started = Instant::now(); + drop(sink); + started.elapsed() + }); + std::thread::sleep(Duration::from_millis(50)); + let drop_time = dropping.join().unwrap(); + assert!(drop_time < Duration::from_millis(600)); + drop(stalled); + } + let deadline = Instant::now() + Duration::from_secs(2); + while Arc::strong_count(&shared) > 1 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(Arc::strong_count(&shared), 1); + drop(shared); + assert_eq!( + std::fs::metadata(&path).unwrap().len(), + header_only_len(directory.path()) + ); + } + + #[test] + fn idle_writer_drop_joins_the_thread() { + let directory = tempfile::tempdir().unwrap(); + let mut sink = MuxSink::Direct(nut_target(&directory.path().join("out.nut"))); + sink.enable_write_behind("test", 1 << 20).unwrap(); + let shared = shared(&sink); + drop(sink); + assert_eq!(Arc::strong_count(&shared), 1); + } + + #[test] + fn writer_panic_is_reported_on_the_next_enqueue() { + let directory = tempfile::tempdir().unwrap(); + let mut target = nut_target(&directory.path().join("out.nut")); + target.panic_on_write = true; + let mut sink = behind(target, 1 << 20, DrainLimits::DEFAULT); + sink.write_packet(packet(0, 64)).unwrap(); + let shared = shared(&sink); + let deadline = Instant::now() + Duration::from_secs(2); + while !lock(&shared.queue).exited && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(sink.write_packet(packet(1024, 64)), Err(ffmpeg::Error::Bug)); + assert_eq!(sink.close(), Err(ffmpeg::Error::Bug)); + } + + #[test] + fn video_budget_scales_with_bitrate_and_machine_memory() { + const GIB: u64 = 1024 * 1024 * 1024; + assert_eq!(video_budget_for(0, Some(16 * GIB)), MIN_VIDEO_BUDGET_BYTES); + assert_eq!(video_budget_for(20_000_000, Some(16 * GIB)), 100_000_000); + assert_eq!( + video_budget_for(u64::MAX, Some(16 * GIB)), + MAX_VIDEO_BUDGET_BYTES + ); + assert_eq!(video_budget_for(u64::MAX, Some(8 * GIB)), 256 * 1024 * 1024); + assert_eq!( + video_budget_for(u64::MAX, Some(GIB)), + MIN_VIDEO_BUDGET_BYTES + ); + assert_eq!(video_budget_for(u64::MAX, None), MAX_VIDEO_BUDGET_BYTES); + } +} From 4ae45b5aff47b053ead0bf8b222696fceee58866 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 14 Oct 2026 18:31:01 +0900 Subject: [PATCH 3/4] fix(recording): move macOS recording file writes off the capture and async paths --- apps/cli/src/selftest/playback.rs | 6 +- .../examples/synthetic-test-runner.rs | 5 +- crates/recording/src/instant_recording.rs | 1 + .../recording/src/output_pipeline/ffmpeg.rs | 115 ++++++++++++++++-- .../output_pipeline/macos_fragmented_m4s.rs | 14 ++- crates/recording/src/studio_recording.rs | 12 +- crates/recording/tests/sync_matrix.rs | 11 +- 7 files changed, 139 insertions(+), 25 deletions(-) diff --git a/apps/cli/src/selftest/playback.rs b/apps/cli/src/selftest/playback.rs index 57e997d9471..4ba3d03f440 100644 --- a/apps/cli/src/selftest/playback.rs +++ b/apps/cli/src/selftest/playback.rs @@ -571,7 +571,7 @@ mod fixture { use cap_recording::{ AudioFrame, ChannelAudioSource, ChannelAudioSourceConfig, ChannelVideoSource, ChannelVideoSourceConfig, OutputPipeline, - ffmpeg::{FFmpegVideoFrame, Mp4Muxer, OggMuxer}, + ffmpeg::{FFmpegVideoFrame, Mp4Muxer, OggMuxer, OggMuxerConfig}, }; use cap_timestamp::{Timestamp, Timestamps}; use relative_path::RelativePathBuf; @@ -783,7 +783,7 @@ mod fixture { audio_info, mic_rx, )) .with_timestamps(timestamps) - .build::(()) + .build::(OggMuxerConfig::default()) .await .map_err(|e| format!("mic pipeline: {e}"))?; let sys_pipeline = OutputPipeline::builder(audio_path.clone()) @@ -794,7 +794,7 @@ mod fixture { // System audio anchors at the recording epoch, exactly like the // studio recorder configures it. .with_audio_anchor(cap_recording::AudioAnchor::PipelineEpoch) - .build::(()) + .build::(OggMuxerConfig::default()) .await .map_err(|e| format!("system audio pipeline: {e}"))?; diff --git a/crates/recording/examples/synthetic-test-runner.rs b/crates/recording/examples/synthetic-test-runner.rs index f2ec2c355e1..8722e2235e5 100644 --- a/crates/recording/examples/synthetic-test-runner.rs +++ b/crates/recording/examples/synthetic-test-runner.rs @@ -1,5 +1,6 @@ use cap_recording::{ - Mp4Muxer, OggMuxer, OutputPipeline, SegmentedVideoMuxer, SegmentedVideoMuxerConfig, + Mp4Muxer, OggMuxer, OggMuxerConfig, OutputPipeline, SegmentedVideoMuxer, + SegmentedVideoMuxerConfig, test_sources::{ AudioGenerator, AudioTestConfig, OutputFormat, RecordingValidator, SyntheticAudioSource, SyntheticAudioSourceConfig, TestConfig, TestPattern, TestPatternVideoSource, @@ -468,7 +469,7 @@ async fn run_audio_only_ogg_recording( let pipeline = OutputPipeline::builder(output_path.to_path_buf()) .with_timestamps(timestamps) .with_audio_source::(audio_source_config) - .build::(()) + .build::(OggMuxerConfig::default()) .await?; tokio::time::sleep(duration + Duration::from_millis(500)).await; diff --git a/crates/recording/src/instant_recording.rs b/crates/recording/src/instant_recording.rs index 759025f1f01..b0faf8aee8b 100644 --- a/crates/recording/src/instant_recording.rs +++ b/crates/recording/src/instant_recording.rs @@ -810,6 +810,7 @@ async fn create_pipeline( output_pipeline::DashSegmentedAudioMuxerConfig { shared_pause_state: None, segment_tx: segment_tx_for_audio, + write_behind: cfg!(target_os = "macos"), ..Default::default() }, ) diff --git a/crates/recording/src/output_pipeline/ffmpeg.rs b/crates/recording/src/output_pipeline/ffmpeg.rs index 1c2f6cbd0e2..26b6793ae4e 100644 --- a/crates/recording/src/output_pipeline/ffmpeg.rs +++ b/crates/recording/src/output_pipeline/ffmpeg.rs @@ -141,11 +141,24 @@ impl AudioMuxer for Mp4Muxer { pub struct OggMuxer(OggFile); +/// `write_behind` moves file writes off the caller (an async runtime worker) +/// onto a dedicated writer thread. Only macOS recordings opt in. +#[derive(Default)] +pub struct OggMuxerConfig { + pub write_behind: bool, +} + +fn warn_inline_fallback(result: std::io::Result<()>) { + if let Err(error) = result { + warn!(%error, "Output write-behind unavailable, muxing inline"); + } +} + impl Muxer for OggMuxer { - type Config = (); + type Config = OggMuxerConfig; async fn setup( - _: Self::Config, + config: Self::Config, output_path: PathBuf, _: Option, audio_config: Option, @@ -158,10 +171,13 @@ impl Muxer for OggMuxer { let audio_config = audio_config.ok_or_else(|| anyhow!("No audio configuration provided"))?; - Ok(Self( - OggFile::init(output_path, |o| OpusEncoder::init(audio_config, o)) - .map_err(|e| anyhow!("Failed to initialize Opus encoder: {e}"))?, - )) + let mut file = OggFile::init(output_path, |o| OpusEncoder::init(audio_config, o)) + .map_err(|e| anyhow!("Failed to initialize Opus encoder: {e}"))?; + if config.write_behind { + warn_inline_fallback(file.enable_write_behind()); + } + + Ok(Self(file)) } fn finish(&mut self, _: Duration) -> anyhow::Result> { @@ -186,6 +202,8 @@ pub struct FragmentedAudioMuxer { #[derive(Default)] pub struct FragmentedAudioMuxerConfig { pub shared_pause_state: Option, + /// See [`OggMuxerConfig::write_behind`]. + pub write_behind: bool, } impl Muxer for FragmentedAudioMuxer { @@ -205,9 +223,14 @@ impl Muxer for FragmentedAudioMuxer { let audio_config = audio_config.ok_or_else(|| anyhow!("No audio configuration provided"))?; + let mut encoder = FragmentedAudioFile::init(output_path, audio_config) + .map_err(|e| anyhow!("Failed to initialize fragmented audio encoder: {e}"))?; + if config.write_behind { + warn_inline_fallback(encoder.enable_write_behind()); + } + Ok(Self { - encoder: FragmentedAudioFile::init(output_path, audio_config) - .map_err(|e| anyhow!("Failed to initialize fragmented audio encoder: {e}"))?, + encoder, pause: config.shared_pause_state, }) } @@ -817,6 +840,8 @@ pub struct DashSegmentedAudioMuxerConfig { pub segment_duration: Duration, pub shared_pause_state: Option, pub segment_tx: Option>, + /// See [`OggMuxerConfig::write_behind`]. + pub write_behind: bool, } impl Default for DashSegmentedAudioMuxerConfig { @@ -825,6 +850,7 @@ impl Default for DashSegmentedAudioMuxerConfig { segment_duration: Duration::from_secs(3), shared_pause_state: None, segment_tx: None, + write_behind: false, } } } @@ -854,6 +880,9 @@ impl Muxer for DashSegmentedAudioMuxer { }, ) .map_err(|e| anyhow!("Failed to initialize DASH audio segment encoder: {e}"))?; + if config.write_behind { + warn_inline_fallback(encoder.enable_write_behind()); + } if let Some(tx) = config.segment_tx { encoder.set_segment_callback(tx); @@ -899,6 +928,76 @@ mod tests { use cap_timestamp::Timestamp; use std::time::Instant; + mod write_behind_opt_in { + use super::super::{ + DashSegmentedAudioMuxer, DashSegmentedAudioMuxerConfig, FragmentedAudioMuxer, + FragmentedAudioMuxerConfig, OggMuxer, OggMuxerConfig, + }; + use crate::{Muxer, TaskPool}; + use cap_media_info::AudioInfo; + use std::sync::{Arc, atomic::AtomicBool}; + + fn audio_info() -> AudioInfo { + AudioInfo::new_raw( + ffmpeg::format::Sample::F32(ffmpeg::format::sample::Type::Packed), + 48_000, + 1, + ) + } + + async fn setup(config: M::Config, path: std::path::PathBuf) -> M { + M::setup( + config, + path, + None, + Some(audio_info()), + Arc::new(AtomicBool::new(false)), + &mut TaskPool::default(), + ) + .await + .unwrap() + } + + #[test] + fn audio_muxers_write_inline_unless_the_call_site_opts_in() { + assert!(!FragmentedAudioMuxerConfig::default().write_behind); + assert!(!DashSegmentedAudioMuxerConfig::default().write_behind); + assert!(!OggMuxerConfig::default().write_behind); + } + + #[tokio::test] + async fn audio_muxers_honour_the_write_behind_flag() { + ffmpeg::init().unwrap(); + let directory = tempfile::tempdir().unwrap(); + for write_behind in [false, true] { + let dir = directory.path().join(write_behind.to_string()); + let fragmented = setup::( + FragmentedAudioMuxerConfig { + shared_pause_state: None, + write_behind, + }, + dir.join("audio.m4a"), + ) + .await; + assert_eq!(fragmented.encoder.is_write_behind(), write_behind); + + let dash = setup::( + DashSegmentedAudioMuxerConfig { + write_behind, + ..Default::default() + }, + dir.join("audio"), + ) + .await; + assert_eq!(dash.encoder.is_write_behind(), write_behind); + + let ogg = + setup::(OggMuxerConfig { write_behind }, dir.join("audio.ogg")).await; + assert_eq!(ogg.0.is_write_behind(), write_behind); + } + } + } + #[test] fn segmented_encoder_preserves_frames_when_a_full_queue_recovers() { let (sender, receiver) = flume::bounded(1); diff --git a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs index 6c490b64e6f..52a2ca06d40 100644 --- a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs @@ -295,8 +295,11 @@ 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_write_behind( + self.base_path.clone(), + self.video_config, + encoder_config, + )?; if let Some(callback) = &self.disk_space_callback { encoder.set_disk_space_callback(callback.clone()); } @@ -682,8 +685,11 @@ 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_write_behind( + self.base_path.clone(), + self.video_config, + encoder_config, + )?; if let Some(callback) = &self.disk_space_callback { encoder.set_disk_space_callback(callback.clone()); } diff --git a/crates/recording/src/studio_recording.rs b/crates/recording/src/studio_recording.rs index 4917df97300..8730061e08d 100644 --- a/crates/recording/src/studio_recording.rs +++ b/crates/recording/src/studio_recording.rs @@ -16,7 +16,7 @@ use crate::{ CursorActor, CursorActorResponse, Cursors, IncrementalCaptureOutputs, spawn_cursor_recorder, }, feeds::{camera::CameraFeedLock, microphone::MicrophoneFeedLock}, - ffmpeg::{FragmentedAudioMuxer, FragmentedAudioMuxerConfig, OggMuxer}, + ffmpeg::{FragmentedAudioMuxer, FragmentedAudioMuxerConfig, OggMuxer, OggMuxerConfig}, output_pipeline::{ AudioAnchor, AudioGapSummary, DoneFut, FinishedOutputPipeline, OutputPipeline, PipelineDoneError, @@ -3436,6 +3436,7 @@ async fn create_segment_pipeline( .with_start_gate(start_gate.clone()) .build::(FragmentedAudioMuxerConfig { shared_pause_state: shared_pause_state.clone(), + write_behind: cfg!(target_os = "macos"), }) .instrument(error_span!("mic-out")) .await @@ -3444,7 +3445,9 @@ async fn create_segment_pipeline( .with_audio_source::(mic_feed) .with_timestamps(start_time) .with_start_gate(start_gate.clone()) - .build::(()) + .build::(OggMuxerConfig { + write_behind: cfg!(target_os = "macos"), + }) .instrument(error_span!("mic-out")) .await }; @@ -3468,6 +3471,7 @@ async fn create_segment_pipeline( .with_audio_anchor(AudioAnchor::PipelineEpoch) .build::(FragmentedAudioMuxerConfig { shared_pause_state: shared_pause_state.clone(), + write_behind: cfg!(target_os = "macos"), }) .instrument(error_span!("system-audio-out")) .await @@ -3477,7 +3481,9 @@ async fn create_segment_pipeline( .with_timestamps(start_time) .with_start_gate(start_gate.clone()) .with_audio_anchor(AudioAnchor::PipelineEpoch) - .build::(()) + .build::(OggMuxerConfig { + write_behind: cfg!(target_os = "macos"), + }) .instrument(error_span!("system-audio-out")) .await }; diff --git a/crates/recording/tests/sync_matrix.rs b/crates/recording/tests/sync_matrix.rs index f321266bdcc..1db05702204 100644 --- a/crates/recording/tests/sync_matrix.rs +++ b/crates/recording/tests/sync_matrix.rs @@ -22,7 +22,8 @@ use cap_recording::{ AudioFrame, ChannelAudioSource, ChannelAudioSourceConfig, ChannelVideoSource, ChannelVideoSourceConfig, OutputPipeline, ffmpeg::{ - FFmpegVideoFrame, Mp4Muxer, OggMuxer, SegmentedVideoMuxer, SegmentedVideoMuxerConfig, + FFmpegVideoFrame, Mp4Muxer, OggMuxer, OggMuxerConfig, SegmentedVideoMuxer, + SegmentedVideoMuxerConfig, }, }; use cap_timestamp::{Timestamp, Timestamps}; @@ -767,7 +768,7 @@ async fn run_audio_case(case: AudioCase) -> Result { let pipeline = OutputPipeline::builder(out_path.clone()) .with_audio_source::(ChannelAudioSourceConfig::new(info, rx)) .with_timestamps(timestamps) - .build::(()) + .build::(OggMuxerConfig::default()) .await .map_err(|e| format!("pipeline build: {e}"))?; @@ -996,7 +997,7 @@ async fn run_system_audio_case(case: SystemAudioCase) -> Result .with_audio_source::(ChannelAudioSourceConfig::new(info, rx)) .with_timestamps(timestamps) .with_audio_anchor(AudioAnchor::PipelineEpoch) - .build::(()) + .build::(OggMuxerConfig::default()) .await .map_err(|e| format!("pipeline build: {e}"))?; @@ -1078,14 +1079,14 @@ async fn run_mic_with_system_audio_case() -> Result { let mic_pipeline = OutputPipeline::builder(mic_path.clone()) .with_audio_source::(ChannelAudioSourceConfig::new(info, mic_rx)) .with_timestamps(timestamps) - .build::(()) + .build::(OggMuxerConfig::default()) .await .map_err(|e| format!("mic pipeline build: {e}"))?; let sys_pipeline = OutputPipeline::builder(sys_path.clone()) .with_audio_source::(ChannelAudioSourceConfig::new(info, sys_rx)) .with_timestamps(timestamps) .with_audio_anchor(AudioAnchor::PipelineEpoch) - .build::(()) + .build::(OggMuxerConfig::default()) .await .map_err(|e| format!("system pipeline build: {e}"))?; From 0db83e2f8a8aeae020fdd58b4db8edbd0af4f112 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Wed, 14 Oct 2026 19:08:40 +0900 Subject: [PATCH 4/4] fix(enc-ffmpeg): bound write-behind finalization and keep segment callbacks off the writer lock --- crates/enc-ffmpeg/src/mux/dash_audio.rs | 46 +- crates/enc-ffmpeg/src/mux/fragmented_audio.rs | 20 +- crates/enc-ffmpeg/src/mux/ogg.rs | 11 +- crates/enc-ffmpeg/src/mux/segmented_stream.rs | 79 +++- crates/enc-ffmpeg/src/mux/write_behind.rs | 423 +++++++++++------- 5 files changed, 373 insertions(+), 206 deletions(-) diff --git a/crates/enc-ffmpeg/src/mux/dash_audio.rs b/crates/enc-ffmpeg/src/mux/dash_audio.rs index 83bfb4f3036..9fef8384949 100644 --- a/crates/enc-ffmpeg/src/mux/dash_audio.rs +++ b/crates/enc-ffmpeg/src/mux/dash_audio.rs @@ -1,6 +1,8 @@ use super::fragment_metadata::read_fragment_metadata; use crate::audio::aac::{AACEncoder, AACEncoderError}; -use crate::mux::segmented_stream::{SegmentCompletedEvent, SegmentMediaType}; +use crate::mux::segmented_stream::{ + SegmentCompletedEvent, SegmentMediaType, SegmentSender, notify_segment, +}; use crate::mux::write_behind::{AUDIO_BUDGET_BYTES, MuxSink, MuxTarget}; use cap_media_info::AudioInfo; use ffmpeg::{format, frame}; @@ -49,6 +51,7 @@ pub struct DashAudioSegmentEncoder { base_path: PathBuf, encoder: AACEncoder, mux: MuxSink, + segment_tx: SegmentSender, } struct AudioSegmentMux { @@ -62,7 +65,7 @@ struct AudioSegmentMux { frames_since_segment_scan: u32, - segment_tx: Option>, + segment_tx: SegmentSender, init_notified: bool, } @@ -230,9 +233,10 @@ impl DashAudioSegmentEncoder { current_index: 1, completed_segments: Vec::new(), frames_since_segment_scan: 0, - segment_tx: None, + segment_tx: SegmentSender::default(), init_notified: false, }; + let segment_tx = mux.segment_tx.clone(); mux.write_in_progress_manifest(); @@ -240,6 +244,7 @@ impl DashAudioSegmentEncoder { base_path, encoder, mux: MuxSink::Direct(mux), + segment_tx, }) } @@ -256,10 +261,14 @@ impl DashAudioSegmentEncoder { } pub fn set_segment_callback(&mut self, tx: std::sync::mpsc::Sender) { - self.mux.with_target(|mux| { - mux.segment_tx = Some(tx); - mux.try_notify_init_segment(); - }); + *self + .segment_tx + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(tx); + if !self.mux.is_write_behind() { + self.mux + .with_target(AudioSegmentMux::try_notify_init_segment); + } } pub fn queue_frame( @@ -277,14 +286,17 @@ impl DashAudioSegmentEncoder { pub fn finish(&mut self) -> Result<(), FinishError> { let encoder = &mut self.encoder; - let flush_result = self - .mux - .finish_flush(|mux| encoder.flush_to(mux)) + self.mux + .finish( + |mux| encoder.flush_to(mux), + |mux: &mut AudioSegmentMux, flush_result| { + let flush_result = flush_result.inspect_err(|error| { + tracing::warn!(%error, "Audio encoder flush failed"); + }); + mux.finish(flush_result) + }, + ) .map_err(std::io::Error::from)? - .inspect_err(|error| { - tracing::warn!(%error, "Audio encoder flush failed"); - }); - self.mux.with_target(|mux| mux.finish(flush_result)) } pub fn finish_with_timestamp(&mut self, _timestamp: Duration) -> Result<(), FinishError> { @@ -327,11 +339,7 @@ impl AudioSegmentMux { } fn notify_segment(&self, event: SegmentCompletedEvent) { - if let Some(tx) = &self.segment_tx - && let Err(e) = tx.send(event) - { - tracing::warn!("Failed to send audio segment completed event: {e}"); - } + notify_segment(&self.segment_tx, event); } fn flush_pending_segments(&mut self) { diff --git a/crates/enc-ffmpeg/src/mux/fragmented_audio.rs b/crates/enc-ffmpeg/src/mux/fragmented_audio.rs index a6d575f2930..c09f61b18d5 100644 --- a/crates/enc-ffmpeg/src/mux/fragmented_audio.rs +++ b/crates/enc-ffmpeg/src/mux/fragmented_audio.rs @@ -144,17 +144,19 @@ impl FragmentedAudioFile { let has_frames = self.has_frames; let encoder = &mut self.encoder; - let flush_result = self + let (flush_result, trailer_result) = self .output - .finish_flush(|output| { - if has_frames { - encoder.flush_to(output) - } else { - Ok(()) - } - }) + .finish( + |output| { + if has_frames { + encoder.flush_to(output) + } else { + Ok(()) + } + }, + |output: &mut AudioOutput, flush_result| (flush_result, output.0.write_trailer()), + ) .map_err(|stalled| FinishError::WriteTrailerFailed(stalled.into()))?; - let trailer_result = self.output.with_target(|output| output.0.write_trailer()); finish_result(flush_result, trailer_result) } } diff --git a/crates/enc-ffmpeg/src/mux/ogg.rs b/crates/enc-ffmpeg/src/mux/ogg.rs index e7a532777d7..e5649d6c8bb 100644 --- a/crates/enc-ffmpeg/src/mux/ogg.rs +++ b/crates/enc-ffmpeg/src/mux/ogg.rs @@ -85,13 +85,14 @@ impl OggFile { self.finished = true; let encoder = &mut self.encoder; - let flush_result = self + let (flush_result, trailer_result) = self .output - .finish_flush(|output| encoder.flush_to(output)) + .finish( + |output| encoder.flush_to(output), + |output: &mut OggOutput, flush_result| (flush_result, output.0.write_trailer()), + ) .map_err(|stalled| FinishError::WriteTrailerFailed(stalled.into()))?; - self.output - .with_target(|output| output.0.write_trailer()) - .map_err(FinishError::WriteTrailerFailed)?; + trailer_result.map_err(FinishError::WriteTrailerFailed)?; Ok(flush_result) } diff --git a/crates/enc-ffmpeg/src/mux/segmented_stream.rs b/crates/enc-ffmpeg/src/mux/segmented_stream.rs index f32d6110ea7..fc72df9d171 100644 --- a/crates/enc-ffmpeg/src/mux/segmented_stream.rs +++ b/crates/enc-ffmpeg/src/mux/segmented_stream.rs @@ -27,6 +27,22 @@ pub struct DiskSpaceWarning { pub type DiskSpaceCallback = Arc; +/// Shared so a write-behind muxer can install the callback without waiting +/// for the writer thread, which may be blocked on storage. +pub(crate) type SegmentSender = + Arc>>>; + +pub(crate) fn notify_segment(sender: &SegmentSender, event: SegmentCompletedEvent) { + let sender = sender + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(tx) = sender.as_ref() + && let Err(e) = tx.send(event) + { + tracing::warn!("Failed to send segment completed event: {e}"); + } +} + fn atomic_write_json(path: &Path, data: &T) -> std::io::Result<()> { let temp_path = path.with_extension("json.tmp"); let json = serde_json::to_string_pretty(data) @@ -78,6 +94,7 @@ pub struct SegmentedVideoEncoder { encoder: H264Encoder, mux: MuxSink, + segment_tx: SegmentSender, segment_duration: Duration, @@ -96,7 +113,7 @@ struct SegmentMux { codec_info: CodecInfo, - segment_tx: Option>, + segment_tx: SegmentSender, init_notified: bool, initial_manifest_pending: bool, } @@ -360,15 +377,17 @@ impl SegmentedVideoEncoder { completed_segments: Vec::new(), frames_since_segment_scan: 0, codec_info, - segment_tx: None, + segment_tx: SegmentSender::default(), init_notified: false, initial_manifest_pending: true, }; + let segment_tx = mux.segment_tx.clone(); let mut instance = Self { base_path, encoder, mux: MuxSink::Direct(mux), + segment_tx, segment_duration: config.segment_duration, disk_space_callback: None, }; @@ -394,10 +413,13 @@ impl SegmentedVideoEncoder { } pub fn set_segment_callback(&mut self, tx: std::sync::mpsc::Sender) { - self.mux.with_target(|mux| { - mux.segment_tx = Some(tx); - mux.try_notify_init_segment(); - }); + *self + .segment_tx + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(tx); + if !self.mux.is_write_behind() { + self.mux.with_target(SegmentMux::try_notify_init_segment); + } } /// Whether frames should be queued as CVPixelBuffers via @@ -443,14 +465,17 @@ impl SegmentedVideoEncoder { pub fn finish(&mut self) -> Result<(), FinishError> { let encoder = &mut self.encoder; - let flush_result = self - .mux - .finish_flush(|mux| encoder.flush_to(mux)) + self.mux + .finish( + |mux| encoder.flush_to(mux), + |mux: &mut SegmentMux, flush_result| { + let flush_result = flush_result.inspect_err(|error| { + tracing::warn!(%error, "Video encoder flush failed"); + }); + mux.finish(flush_result) + }, + ) .map_err(std::io::Error::from)? - .inspect_err(|error| { - tracing::warn!(%error, "Video encoder flush failed"); - }); - self.mux.with_target(|mux| mux.finish(flush_result)) } pub fn finish_with_timestamp(&mut self, _timestamp: Duration) -> Result<(), FinishError> { @@ -545,11 +570,7 @@ impl SegmentMux { } fn notify_segment(&self, event: SegmentCompletedEvent) { - if let Some(tx) = &self.segment_tx - && let Err(e) = tx.send(event) - { - tracing::warn!("Failed to send segment completed event: {e}"); - } + notify_segment(&self.segment_tx, event); } fn flush_pending_segments(&mut self) { @@ -1001,6 +1022,28 @@ mod tests { drop(encoder); } + #[test] + fn write_behind_segment_callback_does_not_wait_for_stalled_storage() { + ffmpeg::init().unwrap(); + let temp = tempfile::tempdir().unwrap(); + let mut encoder = SegmentedVideoEncoder::init_write_behind( + temp.path().to_path_buf(), + test_video_info(), + SegmentedVideoEncoderConfig::default(), + ) + .unwrap(); + encoder.mux.drain(); + let stall = encoder.mux.stall_storage_for(Duration::from_millis(800)); + let started = std::time::Instant::now(); + let (tx, _rx) = mpsc::channel(); + encoder.set_segment_callback(tx); + encoder + .queue_frame(create_test_frame(320, 240), Duration::ZERO) + .unwrap(); + assert!(started.elapsed() < Duration::from_millis(400)); + stall.join().unwrap(); + } + #[test] fn write_behind_publishes_the_same_segments_as_inline_muxing() { ffmpeg::init().unwrap(); diff --git a/crates/enc-ffmpeg/src/mux/write_behind.rs b/crates/enc-ffmpeg/src/mux/write_behind.rs index d2260cf275b..d0155f1c48f 100644 --- a/crates/enc-ffmpeg/src/mux/write_behind.rs +++ b/crates/enc-ffmpeg/src/mux/write_behind.rs @@ -8,16 +8,16 @@ //! budget; once it is full the producer blocks until the writer catches up, //! which is the inline behaviour. //! -//! Finishing drains the queue while storage keeps making progress, bounded -//! by [`DRAIN_NO_PROGRESS_LIMIT`] without a completed write and -//! [`DRAIN_TOTAL_LIMIT`] overall. Past the total limit the unwritten tail is -//! discarded and the output is finalized normally ([`DrainOutcome::Truncated`]). -//! When no write completes for the no-progress limit, finishing returns -//! [`DrainOutcome::Stuck`] without the trailer or final manifest, leaving the -//! files in the state recovery expects from an interrupted recording. As with -//! an encoder blocked in an inline write, the queued packets still reach disk -//! once storage recovers. Dropping a writer that was never finished discards -//! its queue instead and touches nothing after its current call returns. +//! Finishing is one bounded operation: the writer drains the queue, then +//! runs the finalization (trailer, final manifest) itself. When storage is +//! still accepting writes after [`FINISH_DRAIN_LIMIT`], the packets still +//! queued are discarded so finalization can start. If finalization has not +//! completed within [`FINISH_LIMIT`], finishing +//! returns [`StorageStalled`]: finalization is cancelled, the files keep the +//! state recovery expects from an interrupted recording, and the packets +//! still queued reach disk once storage recovers, as they would from an +//! encoder blocked in an inline write. Dropping a writer that was never +//! finished discards its queue and touches nothing after its current call. use crate::base::PacketSink; use ffmpeg::{Packet, Rational, format}; @@ -26,6 +26,7 @@ use std::{ sync::{ Arc, Condvar, Mutex, MutexGuard, atomic::{AtomicBool, Ordering}, + mpsc, }, thread::JoinHandle, time::{Duration, Instant}, @@ -39,10 +40,11 @@ pub(crate) const AUDIO_BUDGET_BYTES: usize = 16 * 1024 * 1024; const STALL_LOG_THRESHOLD: Duration = Duration::from_secs(1); const WRITE_ERROR_LOG_INTERVAL: u64 = 100; // Muxers join their encoder thread for up to 5 s before finishing, and an -// unjoined pipeline stop gives up after 10 s; draining for at most 4 s keeps -// the trailer and manifest inside that window. -pub(crate) const DRAIN_NO_PROGRESS_LIMIT: Duration = Duration::from_secs(4); -pub(crate) const DRAIN_TOTAL_LIMIT: Duration = Duration::from_secs(4); +// unjoined pipeline stop gives up after 10 s; finishing within 4 s keeps the +// whole stop inside that window. +pub(crate) const FINISH_DRAIN_LIMIT: Duration = Duration::from_secs(3); +pub(crate) const FINISH_LIMIT: Duration = Duration::from_secs(4); +const STORAGE_MOVING_WINDOW: Duration = Duration::from_secs(1); const DROP_JOIN_LIMIT: Duration = Duration::from_millis(250); const WAIT_POLL: Duration = Duration::from_millis(50); @@ -98,13 +100,6 @@ impl From for ffmpeg::Error { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum DrainOutcome { - Complete, - Truncated { packets: usize, bytes: usize }, - Stuck, -} - pub(crate) enum MuxSink { Direct(M), Behind(WriteBehind), @@ -123,7 +118,7 @@ impl MuxSink { let Self::Direct(target) = std::mem::replace(self, Self::Detached) else { unreachable!("checked above"); }; - match WriteBehind::spawn(label, target, budget, DrainLimits::DEFAULT) { + match WriteBehind::spawn(label, target, budget, FinishLimits::DEFAULT) { Ok(writer) => { *self = Self::Behind(writer); Ok(()) @@ -161,37 +156,24 @@ impl MuxSink { } } - /// Marks the following packets as the end-of-stream flush: they bypass - /// the budget and their write errors are reported by [`Self::close`], - /// as an inline flush would report them. - pub fn begin_finish(&mut self) { + /// Runs the encoder's end-of-stream `flush` into this sink, then + /// `finalize` with the flush result, which also carries write errors of + /// the end-of-stream packets as an inline flush would. With write-behind + /// both run within [`FINISH_LIMIT`]; `Err` means storage stalled and the + /// output was left unfinalized. + pub fn finish( + &mut self, + flush: impl FnOnce(&mut Self) -> Result<(), ffmpeg::Error>, + finalize: impl FnOnce(&mut M, Result<(), ffmpeg::Error>) -> R + Send + 'static, + ) -> Result { if let Self::Behind(writer) = self { writer.begin_finish(); } - } - - /// Drains the queued packets within the drain limits. An `Err` is the - /// first write error among the end-of-stream packets. - pub fn close(&mut self) -> Result { - match self { - Self::Behind(writer) => writer.close(), - Self::Direct(_) | Self::Detached => Ok(DrainOutcome::Complete), - } - } - - /// Runs the encoder's end-of-stream `flush` into this sink and drains - /// the writer. `Err` means storage stalled: the output was abandoned and - /// must not be finalized. - pub fn finish_flush( - &mut self, - flush: impl FnOnce(&mut Self) -> Result<(), ffmpeg::Error>, - ) -> Result, StorageStalled> { - self.begin_finish(); let flush_result = flush(self); - match self.close() { - Ok(DrainOutcome::Stuck) => Err(StorageStalled), - Ok(DrainOutcome::Complete | DrainOutcome::Truncated { .. }) => Ok(flush_result), - Err(error) => Ok(flush_result.and(Err(error))), + match self { + Self::Direct(target) => Ok(finalize(target, flush_result)), + Self::Behind(writer) => writer.finish(flush_result, finalize), + Self::Detached => unreachable!("mux sink detached"), } } @@ -201,6 +183,22 @@ impl MuxSink { writer.drain(); } } + + #[cfg(test)] + pub fn stall_storage_for(&self, duration: Duration) -> std::thread::JoinHandle<()> { + let Self::Behind(writer) = self else { + panic!("write-behind not enabled"); + }; + let shared = writer.shared.clone(); + let (held_tx, held_rx) = mpsc::channel(); + let handle = std::thread::spawn(move || { + let _held = lock(&shared.target); + held_tx.send(()).unwrap(); + std::thread::sleep(duration); + }); + held_rx.recv().unwrap(); + handle + } } impl PacketSink for MuxSink { @@ -222,34 +220,42 @@ impl PacketSink for MuxSink { } #[derive(Clone, Copy)] -struct DrainLimits { - no_progress: Duration, +struct FinishLimits { + drain: Duration, total: Duration, } -impl DrainLimits { +impl FinishLimits { const DEFAULT: Self = Self { - no_progress: DRAIN_NO_PROGRESS_LIMIT, - total: DRAIN_TOTAL_LIMIT, + drain: FINISH_DRAIN_LIMIT, + total: FINISH_LIMIT, }; } -struct Queue { - packets: VecDeque<(Packet, bool)>, +type Finalizer = Box) + Send>; + +enum Job { + Packet { packet: Packet, end_of_stream: bool }, + Finalize(Finalizer), +} + +struct Queue { + jobs: VecDeque>, bytes: usize, in_flight: bool, closed: bool, finishing: bool, exited: bool, - completed_writes: u64, + last_write_completed: Option, write_errors: u64, - final_error: Option, + end_of_stream_error: Option, } struct Shared { - queue: Mutex, + queue: Mutex>, changed: Condvar, abandoned: AtomicBool, + finalize_cancelled: AtomicBool, target: Mutex, } @@ -260,25 +266,40 @@ fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { } impl Shared { - fn wait<'a>(&self, queue: MutexGuard<'a, Queue>, timeout: Duration) -> MutexGuard<'a, Queue> { + fn wait<'a>( + &self, + queue: MutexGuard<'a, Queue>, + timeout: Duration, + ) -> MutexGuard<'a, Queue> { self.changed .wait_timeout(queue, timeout) .unwrap_or_else(|poisoned| poisoned.into_inner()) .0 } - fn abandon(&self, queue: &mut Queue) -> (usize, usize) { - self.abandoned.store(true, Ordering::Release); - let packets = queue.packets.len(); - let bytes = queue - .packets - .drain(..) - .map(|(packet, _)| packet.size()) - .sum::(); + fn discard_packets(&self, queue: &mut Queue) -> (usize, usize) { + let mut packets = 0; + let mut bytes = 0; + queue.jobs.retain(|job| match job { + Job::Packet { packet, .. } => { + packets += 1; + bytes += packet.size(); + false + } + Job::Finalize(_) => true, + }); queue.bytes -= bytes; self.changed.notify_all(); (packets, bytes) } + + fn abandon(&self, queue: &mut Queue) { + self.abandoned.store(true, Ordering::Release); + self.finalize_cancelled.store(true, Ordering::Release); + queue.jobs.clear(); + queue.bytes = 0; + self.changed.notify_all(); + } } pub(crate) struct WriteBehind { @@ -286,7 +307,7 @@ pub(crate) struct WriteBehind { time_bases: Vec, budget: usize, label: &'static str, - limits: DrainLimits, + limits: FinishLimits, thread: Option>, } @@ -295,7 +316,7 @@ impl WriteBehind { label: &'static str, target: M, budget: usize, - limits: DrainLimits, + limits: FinishLimits, ) -> Result { let time_bases = target .output() @@ -304,18 +325,19 @@ impl WriteBehind { .collect(); let shared = Arc::new(Shared { queue: Mutex::new(Queue { - packets: VecDeque::new(), + jobs: VecDeque::new(), bytes: 0, in_flight: false, closed: false, finishing: false, exited: false, - completed_writes: 0, + last_write_completed: None, write_errors: 0, - final_error: None, + end_of_stream_error: None, }), changed: Condvar::new(), abandoned: AtomicBool::new(false), + finalize_cancelled: AtomicBool::new(false), target: Mutex::new(target), }); let spawned = std::thread::Builder::new() @@ -377,8 +399,11 @@ impl WriteBehind { queue = self.shared.wait(queue, WAIT_POLL); } queue.bytes += size; - let final_packet = queue.finishing; - queue.packets.push_back((packet, final_packet)); + let end_of_stream = queue.finishing; + queue.jobs.push_back(Job::Packet { + packet, + end_of_stream, + }); drop(queue); self.shared.changed.notify_all(); @@ -396,56 +421,73 @@ impl WriteBehind { #[cfg(test)] fn drain(&self) { let mut queue = lock(&self.shared.queue); - while (!queue.packets.is_empty() || queue.in_flight) && !queue.exited { + while (!queue.jobs.is_empty() || queue.in_flight) && !queue.exited { queue = self.shared.wait(queue, WAIT_POLL); } } - fn close(&mut self) -> Result { + fn finish( + &mut self, + flush_result: Result<(), ffmpeg::Error>, + finalize: impl FnOnce(&mut M, Result<(), ffmpeg::Error>) -> R + Send + 'static, + ) -> Result { let Some(thread) = self.thread.take() else { - return Ok(DrainOutcome::Complete); + return Err(StorageStalled); }; - let started = Instant::now(); - let mut last_progress = started; - let mut queue = lock(&self.shared.queue); - queue.closed = true; + let (result_tx, result_rx) = mpsc::sync_channel(1); + { + let mut queue = lock(&self.shared.queue); + queue + .jobs + .push_back(Job::Finalize(Box::new(move |target, end_of_stream| { + let _ = result_tx.send(finalize(target, flush_result.and(end_of_stream))); + }))); + queue.closed = true; + } self.shared.changed.notify_all(); - let mut seen_writes = queue.completed_writes; - let mut truncated = None; - while !queue.exited { - if queue.completed_writes != seen_writes { - seen_writes = queue.completed_writes; - last_progress = Instant::now(); + + let started = Instant::now(); + let mut truncated = false; + loop { + match result_rx.recv_timeout(WAIT_POLL) { + Ok(result) => { + let _ = thread.join(); + return Ok(result); + } + Err(mpsc::RecvTimeoutError::Disconnected) => { + let _ = thread.join(); + tracing::error!(writer = self.label, "Output writer exited before finishing"); + return Err(StorageStalled); + } + Err(mpsc::RecvTimeoutError::Timeout) => {} + } + let storage_moving = lock(&self.shared.queue) + .last_write_completed + .is_some_and(|completed| completed.elapsed() < STORAGE_MOVING_WINDOW); + if !truncated && storage_moving && started.elapsed() >= self.limits.drain { + truncated = true; + let (packets, bytes) = self.shared.discard_packets(&mut lock(&self.shared.queue)); + if packets > 0 { + tracing::warn!( + writer = self.label, + discarded_packets = packets, + discarded_bytes = bytes, + "Output storage too slow to drain the write-behind buffer; finishing without its tail" + ); + } } - if last_progress.elapsed() >= self.limits.no_progress { + if started.elapsed() >= self.limits.total { + self.shared + .finalize_cancelled + .store(true, Ordering::Release); + let queue = lock(&self.shared.queue); tracing::error!( writer = self.label, - stalled_ms = last_progress.elapsed().as_millis() as u64, - queued_packets = queue.packets.len(), queued_bytes = queue.bytes, - "Output storage made no progress while finishing; leaving the recording for recovery and writing the buffer once storage recovers" + "Output storage stalled while finishing; leaving the recording for recovery and writing the buffer once storage recovers" ); - return Ok(DrainOutcome::Stuck); + return Err(StorageStalled); } - if truncated.is_none() && started.elapsed() >= self.limits.total { - let (packets, bytes) = self.shared.abandon(&mut queue); - tracing::warn!( - writer = self.label, - discarded_packets = packets, - discarded_bytes = bytes, - "Output storage too slow to drain the write-behind buffer; finishing without its tail" - ); - truncated = Some(DrainOutcome::Truncated { packets, bytes }); - } - queue = self.shared.wait(queue, WAIT_POLL); - } - let final_error = queue.final_error.take(); - drop(queue); - let panicked = thread.join().is_err(); - match final_error { - Some(error) => Err(error), - None if panicked => Err(ffmpeg::Error::Bug), - None => Ok(truncated.unwrap_or(DrainOutcome::Complete)), } } } @@ -497,7 +539,7 @@ fn run_writer(shared: &Shared, label: &'static str) { if shared.abandoned.load(Ordering::Acquire) { break None; } - if let Some(next) = queue.packets.pop_front() { + if let Some(next) = queue.jobs.pop_front() { queue.in_flight = true; break Some(next); } @@ -510,10 +552,28 @@ fn run_writer(shared: &Shared, label: &'static str) { .unwrap_or_else(|poisoned| poisoned.into_inner()); } }; - let Some((packet, final_packet)) = next else { + let Some(job) = next else { return; }; + let (packet, end_of_stream) = match job { + Job::Packet { + packet, + end_of_stream, + } => (packet, end_of_stream), + Job::Finalize(finalize) => { + let end_of_stream_error = lock(&shared.queue).end_of_stream_error.take(); + let mut target = lock(&shared.target); + if !shared.finalize_cancelled.load(Ordering::Acquire) { + finalize(&mut target, end_of_stream_error.map_or(Ok(()), Err)); + } + drop(target); + lock(&shared.queue).in_flight = false; + shared.changed.notify_all(); + continue; + } + }; + let size = packet.size(); let started = Instant::now(); let result = { @@ -532,13 +592,13 @@ fn run_writer(shared: &Shared, label: &'static str) { let (backlog, error_count) = { let mut queue = lock(&shared.queue); - queue.bytes -= size; + queue.bytes = queue.bytes.saturating_sub(size); queue.in_flight = false; - queue.completed_writes += 1; + queue.last_write_completed = Some(Instant::now()); if let Some(Err(error)) = &result { queue.write_errors += 1; - if final_packet && queue.final_error.is_none() { - queue.final_error = Some(*error); + if end_of_stream && queue.end_of_stream_error.is_none() { + queue.end_of_stream_error = Some(*error); } } (queue.bytes, queue.write_errors) @@ -625,7 +685,7 @@ mod tests { packet } - fn behind(target: NutTarget, budget: usize, limits: DrainLimits) -> MuxSink { + fn behind(target: NutTarget, budget: usize, limits: FinishLimits) -> MuxSink { MuxSink::Behind( WriteBehind::spawn("test", target, budget, limits) .ok() @@ -646,6 +706,17 @@ mod tests { std::fs::metadata(path).unwrap().len() } + fn finish_with_trailer(sink: &mut MuxSink) -> Result<(), ffmpeg::Error> { + sink.finish( + |_| Ok(()), + |target, result| { + target.output.write_trailer().unwrap(); + result + }, + ) + .unwrap() + } + fn written_pts(path: &std::path::Path) -> Vec { let mut input = format::input(path).unwrap(); input @@ -671,8 +742,7 @@ mod tests { assert!(started.elapsed() < Duration::from_secs(1)); drop(stalled); - assert_eq!(sink.close(), Ok(DrainOutcome::Complete)); - sink.with_target(|target| target.output.write_trailer().unwrap()); + finish_with_trailer(&mut sink).unwrap(); assert_eq!( written_pts(&path), (0..64).map(|pts| pts * 1024).collect::>() @@ -699,7 +769,7 @@ mod tests { drop(stalled); let (mut sink, waited) = producer.join().unwrap(); assert!(waited >= Duration::from_millis(250)); - assert_eq!(sink.close(), Ok(DrainOutcome::Complete)); + finish_with_trailer(&mut sink).unwrap(); } #[test] @@ -708,7 +778,7 @@ mod tests { let mut sink = MuxSink::Direct(nut_target(&directory.path().join("out.nut"))); sink.enable_write_behind("test", 1024).unwrap(); sink.write_packet(packet(0, 4096)).unwrap(); - assert_eq!(sink.close(), Ok(DrainOutcome::Complete)); + finish_with_trailer(&mut sink).unwrap(); } #[test] @@ -722,11 +792,17 @@ mod tests { sink.write_packet(invalid_packet(1024)).unwrap(); sink.drain(); sink.write_packet(packet(2048, 64)).unwrap(); - sink.begin_finish(); - sink.write_packet(packet(3072, 64)).unwrap(); - - assert_eq!(sink.close(), Ok(DrainOutcome::Complete)); - sink.with_target(|target| target.output.write_trailer().unwrap()); + let result = sink + .finish( + |sink| sink.write_packet(packet(3072, 64)), + |target, result| { + target.output.write_trailer().unwrap(); + result + }, + ) + .unwrap(); + + assert_eq!(result, Ok(())); assert_eq!(written_pts(&path), vec![0, 2048, 3072]); } @@ -736,24 +812,28 @@ mod tests { let mut sink = MuxSink::Direct(nut_target(&directory.path().join("out.nut"))); sink.enable_write_behind("test", 1 << 20).unwrap(); sink.write_packet(packet(0, 64)).unwrap(); - sink.begin_finish(); - sink.write_packet(invalid_packet(1024)).unwrap(); - assert!(sink.close().is_err()); + let result = sink + .finish( + |sink| sink.write_packet(invalid_packet(1024)), + |_, result| result, + ) + .unwrap(); + assert!(result.is_err()); } #[test] - fn finish_returns_on_stuck_storage_and_keeps_the_buffer_for_when_it_recovers() { + fn stalled_finish_returns_in_time_and_keeps_the_buffer_for_when_storage_recovers() { let directory = tempfile::tempdir().unwrap(); let path = directory.path().join("out.nut"); let packets = 8usize; let packet_size = 64 * 1024; + let finalized = Arc::new(AtomicBool::new(false)); { - let target = nut_target(&path); - let limits = DrainLimits { - no_progress: Duration::from_millis(200), - total: Duration::from_secs(30), + let limits = FinishLimits { + drain: Duration::from_millis(100), + total: Duration::from_millis(300), }; - let mut sink = behind(target, 1 << 20, limits); + let mut sink = behind(nut_target(&path), 1 << 20, limits); let shared = shared(&sink); let stalled = lock(&shared.target); for pts in 0..packets { @@ -761,8 +841,13 @@ mod tests { .unwrap(); } let started = Instant::now(); - assert_eq!(sink.close(), Ok(DrainOutcome::Stuck)); - assert!(started.elapsed() < Duration::from_secs(2)); + let finalized_flag = finalized.clone(); + let result = sink.finish( + |_| Ok(()), + move |_, _| finalized_flag.store(true, Ordering::Release), + ); + assert_eq!(result, Err(StorageStalled)); + assert!(started.elapsed() < Duration::from_secs(1)); drop(sink); drop(stalled); let deadline = Instant::now() + Duration::from_secs(2); @@ -771,6 +856,7 @@ mod tests { } assert_eq!(Arc::strong_count(&shared), 1); } + assert!(!finalized.load(Ordering::Acquire)); assert!( std::fs::metadata(&path).unwrap().len() >= header_only_len(directory.path()) + (packets * packet_size) as u64 @@ -778,32 +864,34 @@ mod tests { } #[test] - fn slow_storage_drains_within_the_total_limit_then_finishes_without_the_tail() { + fn slow_storage_finishes_without_the_tail_within_the_limit() { let directory = tempfile::tempdir().unwrap(); let path = directory.path().join("out.nut"); let mut target = nut_target(&path); target.write_delay = Duration::from_millis(60); - let limits = DrainLimits { - no_progress: Duration::from_secs(2), - total: Duration::from_millis(300), + let limits = FinishLimits { + drain: Duration::from_millis(300), + total: Duration::from_secs(2), }; let mut sink = behind(target, 1 << 20, limits); for pts in 0..40 { sink.write_packet(packet(pts * 1024, 64)).unwrap(); } let started = Instant::now(); - let outcome = sink.close().unwrap(); + let result = sink + .finish( + |_| Ok(()), + |target, result| { + target.write_delay = Duration::ZERO; + target.output.write_trailer().unwrap(); + result + }, + ) + .unwrap(); + assert_eq!(result, Ok(())); assert!(started.elapsed() < Duration::from_secs(2)); - let DrainOutcome::Truncated { packets, .. } = outcome else { - panic!("expected a truncated drain, got {outcome:?}"); - }; - assert!(packets > 0); - sink.with_target(|target| { - target.write_delay = Duration::ZERO; - target.output.write_trailer().unwrap(); - }); let pts = written_pts(&path); - assert_eq!(pts.len() + packets, 40); + assert!(!pts.is_empty() && pts.len() < 40); assert_eq!( pts, (0..pts.len() as i64) @@ -812,6 +900,28 @@ mod tests { ); } + #[test] + fn slow_finalization_counts_against_the_finish_limit() { + let directory = tempfile::tempdir().unwrap(); + let limits = FinishLimits { + drain: Duration::from_millis(100), + total: Duration::from_millis(300), + }; + let mut sink = behind( + nut_target(&directory.path().join("out.nut")), + 1 << 20, + limits, + ); + sink.write_packet(packet(0, 64)).unwrap(); + let started = Instant::now(); + let result = sink.finish( + |_| Ok(()), + |_, _| std::thread::sleep(Duration::from_secs(1)), + ); + assert_eq!(result, Err(StorageStalled)); + assert!(started.elapsed() < Duration::from_millis(900)); + } + #[test] fn dropping_without_finish_stops_the_writer_promptly() { let directory = tempfile::tempdir().unwrap(); @@ -862,7 +972,7 @@ mod tests { let directory = tempfile::tempdir().unwrap(); let mut target = nut_target(&directory.path().join("out.nut")); target.panic_on_write = true; - let mut sink = behind(target, 1 << 20, DrainLimits::DEFAULT); + let mut sink = behind(target, 1 << 20, FinishLimits::DEFAULT); sink.write_packet(packet(0, 64)).unwrap(); let shared = shared(&sink); let deadline = Instant::now() + Duration::from_secs(2); @@ -870,7 +980,10 @@ mod tests { std::thread::sleep(Duration::from_millis(10)); } assert_eq!(sink.write_packet(packet(1024, 64)), Err(ffmpeg::Error::Bug)); - assert_eq!(sink.close(), Err(ffmpeg::Error::Bug)); + assert_eq!( + sink.finish(|_| Ok(()), |_, result| result), + Err(StorageStalled) + ); } #[test]