From a9ed9484f71f65653ac958b3471551821b41c635 Mon Sep 17 00:00:00 2001 From: sloemodzn Date: Wed, 14 Oct 2026 11:52:08 +0530 Subject: [PATCH 1/3] fix(recording): keep audio through multi-second pipeline stalls Audio channels were 128 frames deep (~2.7 s of 20 ms frames), so a consumer stall longer than that made sources shed frames and the muxer later filled the hole with silence. Recording to slow storage can stall consumers for seconds at a time (#2366). 2048 frames is ~40 s of audio and costs a few tens of MB, small next to what the video pipeline already holds. The system-audio stats task also warned every 5 s for the rest of a recording after a single drop; now it warns only when the drop counter advanced. The microphone send timeout was tight enough (5 ms wired, 20 ms wireless) for scheduler jitter to drop frames even with channel space available, so it moves to 50 ms / 200 ms; the channel itself now absorbs the multi-second stalls. --- crates/recording/src/output_pipeline/core.rs | 35 +++++++++++++++++-- crates/recording/src/sources/microphone.rs | 13 ++++--- .../src/sources/screen_capture/macos.rs | 33 +++++++++-------- 3 files changed, 61 insertions(+), 20 deletions(-) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 9045431f630..75f2d0709dc 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -1698,6 +1698,37 @@ async fn next_frame(rx: &mut S) -> (Option, const DEFAULT_VIDEO_SOURCE_CHANNEL_CAPACITY: usize = 300; +/// Audio frames are small (~8 KB for 20 ms of 48 kHz stereo), so a deep audio +/// channel costs a few tens of MB at most, while shedding frames turns a +/// multi-second source or consumer stall (slow storage, sleep/wake, runtime +/// hiccups) into audible silence. 2048 twenty-millisecond frames is ~40 s. +pub(crate) const AUDIO_SOURCE_CHANNEL_CAPACITY: usize = 2048; + +#[cfg(test)] +mod audio_channel_capacity_tests { + use super::AUDIO_SOURCE_CHANNEL_CAPACITY; + + /// Recording to slow storage can stall the consumer of an audio channel + /// for many seconds. The channel has to absorb that stall, otherwise the + /// source sheds frames and the muxer fills the hole with silence. + #[test] + fn audio_channel_absorbs_multi_second_consumer_stalls() { + let frame_secs = 0.02; + let buffered_secs = AUDIO_SOURCE_CHANNEL_CAPACITY as f64 * frame_secs; + assert!( + buffered_secs >= 30.0, + "audio channel buffers only {buffered_secs:.1} s, too little for multi-second stalls" + ); + + // 20 s of frames queued with nothing draining: every send must fit. + let (tx, _rx) = tokio::sync::mpsc::channel::(AUDIO_SOURCE_CHANNEL_CAPACITY); + for _ in 0..1000 { + tx.try_send(0) + .expect("audio channel shed a frame during a 20 s stall"); + } + } +} + fn get_video_source_channel_capacity() -> usize { std::env::var("CAP_VIDEO_SOURCE_BUFFER_SIZE") .ok() @@ -4820,7 +4851,7 @@ async fn setup_audio_sources( } let mut erased_audio_sources = vec![]; - let (audio_tx, audio_rx) = mpsc::channel(128); + let (audio_tx, audio_rx) = mpsc::channel(AUDIO_SOURCE_CHANNEL_CAPACITY); let audio_info = if audio_sources.len() == 1 { let source = (audio_sources.swap_remove(0))(audio_tx, setup_ctx).await?; @@ -4841,7 +4872,7 @@ async fn setup_audio_sources( let (ready_tx, ready_rx) = oneshot::channel::>(); for audio_source_setup in audio_sources { - let (tx, rx) = mpsc::channel(128); + let (tx, rx) = mpsc::channel(AUDIO_SOURCE_CHANNEL_CAPACITY); let source = (audio_source_setup)(tx, setup_ctx).await?; audio_mixer.add_source(source.audio_info, rx); diff --git a/crates/recording/src/sources/microphone.rs b/crates/recording/src/sources/microphone.rs index a81ea9da8f5..27332a88cab 100644 --- a/crates/recording/src/sources/microphone.rs +++ b/crates/recording/src/sources/microphone.rs @@ -1,6 +1,8 @@ use crate::{ feeds::microphone::{self, MicrophoneFeedLock}, - output_pipeline::{AudioFrame, AudioSource, PipelineHealthEvent, emit_health}, + output_pipeline::{ + AUDIO_SOURCE_CHANNEL_CAPACITY, AudioFrame, AudioSource, PipelineHealthEvent, emit_health, + }, sources::audio_mixer::AudioMixer, }; use cap_media_info::{AudioInfo, ffmpeg_sample_format_for}; @@ -309,12 +311,15 @@ impl AudioSource for Microphone { let cancel = CancellationToken::new(); #[cfg(windows)] let cancel = source_cancel; - let (tx, rx) = flume::bounded(128); + let (tx, rx) = flume::bounded(AUDIO_SOURCE_CHANNEL_CAPACITY); + // The timeout only guards against a stuck feed; the channel itself + // absorbs multi-second consumer stalls, so a few tens of + // milliseconds of scheduler jitter must not drop audio. let send_timeout = if is_wireless { - Duration::from_millis(20) + Duration::from_millis(200) } else { - Duration::from_millis(5) + Duration::from_millis(50) }; #[cfg(windows)] diff --git a/crates/recording/src/sources/screen_capture/macos.rs b/crates/recording/src/sources/screen_capture/macos.rs index 66972d65eb4..b15d1785f03 100644 --- a/crates/recording/src/sources/screen_capture/macos.rs +++ b/crates/recording/src/sources/screen_capture/macos.rs @@ -16,7 +16,7 @@ use std::{ Arc, Mutex, atomic::{self, AtomicBool, AtomicU32, AtomicU64}, }, - time::{Duration, Instant}, + time::Duration, }; use tokio::{select, sync::broadcast}; use tokio_util::{ @@ -208,7 +208,7 @@ impl ScreenCaptureConfig { let (video_tx, video_rx) = flume::bounded(buffer_size); let drop_counter: Arc = Arc::new(AtomicU64::new(0)); let (mut audio_tx, audio_rx) = if self.system_audio { - let (tx, rx) = mpsc::channel(128); + let (tx, rx) = mpsc::channel(output_pipeline::AUDIO_SOURCE_CHANNEL_CAPACITY); (Some(tx), Some(rx)) } else { (None, None) @@ -1263,13 +1263,18 @@ impl output_pipeline::AudioSource for SystemAudioSource { let stats_cancel = cancel_token.clone(); tokio::spawn( async move { - let mut last_log = Instant::now(); + let mut prev_captured: u64 = 0; + let mut prev_dropped: u64 = 0; loop { tokio::time::sleep(Duration::from_secs(5)).await; let captured = frame_counter.load(atomic::Ordering::Relaxed); let dropped = drop_counter.load(atomic::Ordering::Relaxed); - if dropped > 0 { + // Report drops that happened since the last tick only: a + // single hiccup used to keep warning every five seconds + // for the rest of the recording. + let new_drops = dropped.saturating_sub(prev_dropped); + if new_drops > 0 { let total = captured + dropped; let drop_pct = if total > 0 { 100.0 * dropped as f64 / total as f64 @@ -1277,18 +1282,18 @@ impl output_pipeline::AudioSource for SystemAudioSource { 0.0 }; - if last_log.elapsed() >= Duration::from_secs(5) { - warn!( - captured = captured, - dropped = dropped, - drop_pct = format!("{:.1}%", drop_pct), - "System audio dropping frames due to full channel" - ); - last_log = Instant::now(); - } - } else if captured > 0 { + warn!( + captured = captured, + dropped = dropped, + new_drops = new_drops, + drop_pct = format!("{:.1}%", drop_pct), + "System audio dropping frames due to full channel" + ); + } else if captured != prev_captured { debug!(captured = captured, "System audio frames captured"); } + prev_captured = captured; + prev_dropped = dropped; } } .with_cancellation_token_owned(stats_cancel) From 8b4aa3694ffecf5f80ed9de60f0f766a18a75340 Mon Sep 17 00:00:00 2001 From: sloemodzn Date: Wed, 14 Oct 2026 12:21:23 +0530 Subject: [PATCH 2/3] fix(recording): drain the full audio backlog on stop The stop path only saved 500 queued audio frames, so with the enlarged source queue a long stall could still end with discarded samples and silence padding. The drain now runs to the channel capacity. The stall test goes through the production channel factory with real 20 ms frames instead of a private byte queue, so it fails if a production queue shrinks back. --- crates/recording/src/output_pipeline/core.rs | 49 ++++++++++++-------- 1 file changed, 30 insertions(+), 19 deletions(-) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 75f2d0709dc..98eacd302c7 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -1704,28 +1704,39 @@ const DEFAULT_VIDEO_SOURCE_CHANNEL_CAPACITY: usize = 300; /// hiccups) into audible silence. 2048 twenty-millisecond frames is ~40 s. pub(crate) const AUDIO_SOURCE_CHANNEL_CAPACITY: usize = 2048; +fn audio_source_channel() -> (mpsc::Sender, mpsc::Receiver) { + mpsc::channel(AUDIO_SOURCE_CHANNEL_CAPACITY) +} + #[cfg(test)] mod audio_channel_capacity_tests { - use super::AUDIO_SOURCE_CHANNEL_CAPACITY; + use super::*; - /// Recording to slow storage can stall the consumer of an audio channel - /// for many seconds. The channel has to absorb that stall, otherwise the - /// source sheds frames and the muxer fills the hole with silence. #[test] - fn audio_channel_absorbs_multi_second_consumer_stalls() { - let frame_secs = 0.02; - let buffered_secs = AUDIO_SOURCE_CHANNEL_CAPACITY as f64 * frame_secs; - assert!( - buffered_secs >= 30.0, - "audio channel buffers only {buffered_secs:.1} s, too little for multi-second stalls" - ); + fn audio_channel_absorbs_thirty_second_consumer_stalls() { + let timestamps = Timestamps::now(); + // 30 s of 20 ms frames, matching the microphone silence chunk size. + let frames = 1500usize; + let (mut tx, mut rx) = audio_source_channel(); + for _ in 0..frames { + let mut frame = ffmpeg::frame::Audio::new( + ffmpeg::format::Sample::F32(ffmpeg::format::sample::Type::Planar), + 960, + ffmpeg::ChannelLayout::STEREO, + ); + frame.set_rate(48_000); + tx.try_send(AudioFrame::new( + frame, + Timestamp::Instant(timestamps.instant()), + )) + .expect("audio channel shed a frame during a 30 s stall"); + } - // 20 s of frames queued with nothing draining: every send must fit. - let (tx, _rx) = tokio::sync::mpsc::channel::(AUDIO_SOURCE_CHANNEL_CAPACITY); - for _ in 0..1000 { - tx.try_send(0) - .expect("audio channel shed a frame during a 20 s stall"); + let mut recovered = 0usize; + while matches!(rx.try_next(), Ok(Some(_))) { + recovered += 1; } + assert_eq!(recovered, frames); } } @@ -4172,7 +4183,7 @@ impl PreparedAudioSources { if was_cancelled && !audio_degraded { let drain_start = std::time::Instant::now(); let drain_deadline = tokio::time::Instant::now() + MIXER_STOP_TIMEOUT; - let max_drain_frames = 500u64; + let max_drain_frames = AUDIO_SOURCE_CHANNEL_CAPACITY as u64; let mut drained = 0u64; let mut skipped = 0u64; let mut degraded_during_drain = false; @@ -4851,7 +4862,7 @@ async fn setup_audio_sources( } let mut erased_audio_sources = vec![]; - let (audio_tx, audio_rx) = mpsc::channel(AUDIO_SOURCE_CHANNEL_CAPACITY); + let (audio_tx, audio_rx) = audio_source_channel(); let audio_info = if audio_sources.len() == 1 { let source = (audio_sources.swap_remove(0))(audio_tx, setup_ctx).await?; @@ -4872,7 +4883,7 @@ async fn setup_audio_sources( let (ready_tx, ready_rx) = oneshot::channel::>(); for audio_source_setup in audio_sources { - let (tx, rx) = mpsc::channel(AUDIO_SOURCE_CHANNEL_CAPACITY); + let (tx, rx) = audio_source_channel(); let source = (audio_source_setup)(tx, setup_ctx).await?; audio_mixer.add_source(source.audio_info, rx); From 49a8523ec7c13425c712dd0452751982720919e2 Mon Sep 17 00:00:00 2001 From: sloemodzn Date: Wed, 14 Oct 2026 12:42:39 +0530 Subject: [PATCH 3/3] test(recording): fill the combined audio channel to its production capacity The drain-failure test assumed a 128-frame source channel and capped its fill loop at 256 sends. The source channel now buffers AUDIO_SOURCE_CHANNEL_CAPACITY frames, so the loop bound follows the constant instead of a hardcoded count. --- crates/recording/src/instant_recording.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/recording/src/instant_recording.rs b/crates/recording/src/instant_recording.rs index bae7bf0085e..278bc21cef0 100644 --- a/crates/recording/src/instant_recording.rs +++ b/crates/recording/src/instant_recording.rs @@ -1777,9 +1777,9 @@ mod tests { mod quiescence_tests { use super::*; use crate::output_pipeline::{ - AudioFrame, AudioMuxer, AudioSource, ChannelAudioSource, ChannelAudioSourceConfig, - ChannelVideoSource, ChannelVideoSourceConfig, Muxer, SetupCtx, TaskPool, VideoFrame, - VideoMuxer, + AUDIO_SOURCE_CHANNEL_CAPACITY, AudioFrame, AudioMuxer, AudioSource, ChannelAudioSource, + ChannelAudioSourceConfig, ChannelVideoSource, ChannelVideoSourceConfig, Muxer, SetupCtx, + TaskPool, VideoFrame, VideoMuxer, }; use cap_timestamp::Timestamp; use std::{ @@ -2463,7 +2463,7 @@ mod quiescence_tests { .try_send(combined_audio_frame(fixture.timestamps, 0.25)) .unwrap(); let mut full = false; - for _ in 0..256 { + for _ in 0..(AUDIO_SOURCE_CHANNEL_CAPACITY + 8) { if let Err(error) = fixture .audio .try_send(combined_audio_frame(fixture.timestamps, 0.75))