From 65edf214b9801c879a1cd7b3e18a6b3c0ea7244c Mon Sep 17 00:00:00 2001 From: fufesou Date: Wed, 9 Sep 2026 22:34:25 +0800 Subject: [PATCH] fix(macos): recover system-stopped audio capture streams (#16123) * fix(macos): recreate system-stopped audio capture streams Pin CPAL's ScreenCaptureKit stop notifications and retain interruption state with each capture stream. Recreate an interrupted stream through the existing service restart path, outside the backend error callback, and resend its audio format. Late callbacks cannot restart a replacement. A natural -3821 stop was observed with the remote connection still open. Its OS trigger remains unknown and it has no deterministic natural reproducer. Controlled verification stops the real SCStream and delivers an explicitly marked -3821 notification; this is not a natural failure. Dependency: https://github.com/rustdesk-org/cpal/pull/5 Validation: requested macOS Rust and Flutter debug builds; three full-crate regression tests; build check without ScreenCaptureKit; two controlled recreations on one connection with independently recorded receiver audio. * chore(macos): log audio capture startup and resumed samples * Update deps, cpal Signed-off-by: fufesou --------- Signed-off-by: fufesou --- Cargo.lock | 2 +- src/server/audio_capture_error.rs | 72 +++++++++++++++++++++++++++++++ src/server/audio_service.rs | 61 +++++++++++++++++++++----- 3 files changed, 123 insertions(+), 12 deletions(-) create mode 100644 src/server/audio_capture_error.rs diff --git a/Cargo.lock b/Cargo.lock index e58ffd397..f54e14e02 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1717,7 +1717,7 @@ dependencies = [ [[package]] name = "cpal" version = "0.15.3" -source = "git+https://github.com/rustdesk-org/cpal?branch=osx-screencapturekit#41c8a2a75903ffe9cd44cf176ce6b69e5d916d47" +source = "git+https://github.com/rustdesk-org/cpal?branch=osx-screencapturekit#69ad2578adc9200093fc81cdfbdad63dbc4274f9" dependencies = [ "alsa", "cidre", diff --git a/src/server/audio_capture_error.rs b/src/server/audio_capture_error.rs new file mode 100644 index 000000000..bb837d8bf --- /dev/null +++ b/src/server/audio_capture_error.rs @@ -0,0 +1,72 @@ +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; + +// Each stream owns its flag so a late callback cannot restart a replacement. +#[derive(Clone, Default)] +pub(super) struct CaptureErrorHandler { + interrupted: Arc, +} + +impl CaptureErrorHandler { + pub(super) fn handle(&self, error: cpal::StreamError) { + if matches!(error, cpal::StreamError::StreamInterrupted { .. }) { + // ScreenCaptureKit can stop capture while the remote session stays open. + // The observed -3821 error does not identify its underlying trigger. + // https://developer.apple.com/documentation/screencapturekit/scstreamdelegate/stream(_:didstopwitherror:) + hbb_common::log::error!("Audio capture stream interrupted: {error}"); + self.interrupted.store(true, Ordering::Relaxed); + } else { + // Keep frequent sample-buffer errors at the existing trace level. + hbb_common::log::trace!("an error occurred on stream: {error}"); + } + } + + pub(super) fn needs_restart(&self) -> bool { + self.interrupted.load(Ordering::Relaxed) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use cpal::{BackendSpecificError, StreamError}; + + fn system_interruption() -> StreamError { + StreamError::StreamInterrupted { + err: BackendSpecificError { + description: "ScreenCaptureKit system-stopped capture (-3821)".to_owned(), + }, + } + } + + #[test] + fn system_interruption_requests_recreation_of_the_active_stream() { + let errors = CaptureErrorHandler::default(); + let callback = errors.clone(); + assert!(!errors.needs_restart()); + callback.handle(system_interruption()); + assert!(errors.needs_restart()); + } + + #[test] + fn late_error_from_an_old_stream_does_not_restart_its_replacement() { + let old_callback = CaptureErrorHandler::default(); + let replacement = CaptureErrorHandler::default(); + old_callback.handle(system_interruption()); + assert!(old_callback.needs_restart()); + assert!(!replacement.needs_restart()); + } + + #[test] + fn other_backend_errors_do_not_request_recreation() { + let errors = CaptureErrorHandler::default(); + errors.handle(StreamError::BackendSpecific { + err: BackendSpecificError { + description: "A sample buffer could not be read".to_owned(), + }, + }); + assert!(!errors.needs_restart()); + } +} diff --git a/src/server/audio_service.rs b/src/server/audio_service.rs index b58f83bcf..52f2eff2e 100644 --- a/src/server/audio_service.rs +++ b/src/server/audio_service.rs @@ -171,9 +171,14 @@ pub fn is_screen_capture_kit_available() -> bool { .any(|host| *host == cpal::HostId::ScreenCaptureKit) } +#[cfg(not(any(target_os = "linux", target_os = "android")))] +#[path = "audio_capture_error.rs"] +mod audio_capture_error; + #[cfg(not(any(target_os = "linux", target_os = "android")))] mod cpal_impl { use self::service::{Reset, ServiceSwap}; + use super::audio_capture_error::CaptureErrorHandler; use super::*; use cpal::{ traits::{DeviceTrait, HostTrait, StreamTrait}, @@ -192,7 +197,7 @@ mod cpal_impl { #[derive(Default)] pub struct State { - stream: Option<(Box, Arc)>, + stream: Option<(Box, Arc, CaptureErrorHandler)>, } impl super::service::Reset for State { @@ -210,8 +215,10 @@ mod cpal_impl { } _ => {} } - if let Some((_, format)) = &state.stream { + if let Some((_, format, _)) = &state.stream { sp.send_shared(format.clone()); + #[cfg(target_os = "macos")] + log::info!("Audio capture stream recreated; replacement format sent"); } RESTARTING.store(false, Ordering::SeqCst); Ok(()) @@ -225,7 +232,7 @@ mod cpal_impl { } _ => {} } - if let Some((_, format)) = &state.stream { + if let Some((_, format, _)) = &state.stream { sps.send_shared(format.clone()); } Ok(()) @@ -234,6 +241,13 @@ mod cpal_impl { } pub fn run(sp: EmptyExtraFieldService, state: &mut State) -> ResultType<()> { + if let Some((_, _, errors)) = &state.stream { + if errors.needs_restart() { + // Recreate on the service thread, outside the backend's error callback. + log::warn!("Recreating interrupted audio capture stream"); + super::restart(); + } + } if !RESTARTING.load(Ordering::SeqCst) { run_serv_snapshot(sp, state) } else { @@ -353,7 +367,9 @@ mod cpal_impl { Ok((device, format)) } - fn play(sp: &GenericService) -> ResultType<(Box, Arc)> { + fn play( + sp: &GenericService, + ) -> ResultType<(Box, Arc, CaptureErrorHandler)> { use cpal::SampleFormat::*; let (device, config) = get_device()?; let sp = sp.clone(); @@ -371,7 +387,7 @@ mod cpal_impl { 48000 }; let ch = if config.channels() > 1 { Stereo } else { Mono }; - let stream = match config.sample_format() { + let (stream, errors) = match config.sample_format() { I8 => build_input_stream::(device, &config, sp, sample_rate, ch)?, I16 => build_input_stream::(device, &config, sp, sample_rate, ch)?, I32 => build_input_stream::(device, &config, sp, sample_rate, ch)?, @@ -385,9 +401,12 @@ mod cpal_impl { f => bail!("unsupported audio format: {:?}", f), }; stream.play()?; + #[cfg(target_os = "macos")] + log::info!("Audio capture start call succeeded"); Ok(( Box::new(stream), Arc::new(create_format_msg(sample_rate, ch as _)), + errors, )) } @@ -397,14 +416,15 @@ mod cpal_impl { sp: GenericService, sample_rate: u32, encode_channel: magnum_opus::Channels, - ) -> ResultType + ) -> ResultType<(cpal::Stream, CaptureErrorHandler)> where T: cpal::SizedSample + dasp::sample::ToSample, { - let err_fn = move |err| { - // too many UnknownErrno, will improve later - log::trace!("an error occurred on stream: {}", err); - }; + let errors = CaptureErrorHandler::default(); + let callback_errors = errors.clone(); + let err_fn = move |err| callback_errors.handle(err); + #[cfg(target_os = "macos")] + let (mut received_samples, mut received_signal) = (false, false); let sample_rate_0 = config.sample_rate().0; log::debug!("Audio sample rate : {}", sample_rate); unsafe { @@ -431,6 +451,25 @@ mod cpal_impl { &stream_config, move |data: &[T], _: &InputCallbackInfo| { let buffer: Vec = data.iter().map(|s| T::to_sample(*s)).collect(); + #[cfg(target_os = "macos")] + { + // Starting capture does not guarantee sample delivery or audible data. + if !received_samples && !buffer.is_empty() { + received_samples = true; + log::info!( + "Audio capture received first PCM block: {} samples", + buffer.len() + ); + } + if !received_signal + && buffer + .iter() + .any(|sample| sample.is_finite() && *sample != 0.0) + { + received_signal = true; + log::info!("Audio capture received first nonzero PCM"); + } + } let mut lock = INPUT_BUFFER.lock().unwrap(); lock.extend(buffer); while lock.len() >= rechannel_len { @@ -449,7 +488,7 @@ mod cpal_impl { err_fn, timeout, )?; - Ok(stream) + Ok((stream, errors)) } }