fix(recording): keep audio through multi-second pipeline stalls

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))

crates/recording/src/output_pipeline/core.rs

@@ -1698,6 +1698,48 @@ async fn next_frame<S: futures::Stream + Unpin>(rx: &mut S) -> (Option<S::Item>,
 
 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<AudioFrame>, mpsc::Receiver<AudioFrame>) {
+    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::<anyhow::Result<()>>();
 
         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);

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)]

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<CMSampleBufferCapture> {
         let (video_tx, video_rx) = flume::bounded(buffer_size);
         let drop_counter: Arc<AtomicU64> = 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,32 +1263,37 @@ 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
                         } else {
                             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)