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)) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 9045431f630..98eacd302c7 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -1698,6 +1698,48 @@ 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; + +fn audio_source_channel() -> (mpsc::Sender, mpsc::Receiver) { + mpsc::channel(AUDIO_SOURCE_CHANNEL_CAPACITY) +} + +#[cfg(test)] +mod audio_channel_capacity_tests { + use super::*; + + #[test] + 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"); + } + + let mut recovered = 0usize; + while matches!(rx.try_next(), Ok(Some(_))) { + recovered += 1; + } + assert_eq!(recovered, frames); + } +} + fn get_video_source_channel_capacity() -> usize { std::env::var("CAP_VIDEO_SOURCE_BUFFER_SIZE") .ok() @@ -4141,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; @@ -4820,7 +4862,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) = audio_source_channel(); let audio_info = if audio_sources.len() == 1 { let source = (audio_sources.swap_remove(0))(audio_tx, setup_ctx).await?; @@ -4841,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(128); + let (tx, rx) = audio_source_channel(); 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)