fix(audio): preserve compatible playback and recover failed Windows outputs (#16150)

* fix(audio): preserve compatible playback on replacement failure

* fix(audio): use Windows events for capture worker wakeups

* fix(audio): recover Windows playback after output device failure

* reduce diffs

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): reset decoder when retaining compatible playback

* fix(audio): integrate playback retention and Windows capture wakeups

* revert(audio): remove Windows capture-event notification changes

* docs(audio): explain Windows 7 capture limitations

* docs(audio): clarify capture changes introduced by #16095

* fix(audio): preserve playback through asynchronous Windows startup

Keep a compatible active output until the replacement callback confirms startup, preserving decoder progress on promotion or rollback. Handle superseding formats and simultaneous output failures without losing recovery state. Pin the scoped CPAL WASAPI event-ownership fix and add deterministic and native regressions.

* fix(audio): retain ready output across superseding formats

* fix(audio): update CPAL teardown recovery

Pin the upstream-aligned WASAPI cleanup with shared event lifetime, self-join prevention and fallible destructor diagnostics. Preserve the existing dependency graph and audio implementation.

* Use upstream-style CPAL stream teardown

* Refact: remove low value test

Signed-off-by: fufesou <linlong1266@gmail.com>

* fix(audio): retry playback when startup confirmation times out

Signed-off-by: fufesou <linlong1266@gmail.com>

* update cpal

Signed-off-by: fufesou <linlong1266@gmail.com>

---------

Signed-off-by: fufesou <linlong1266@gmail.com>

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#69ad2578adc9200093fc81cdfbdad63dbc4274f9"
+source = "git+https://github.com/rustdesk-org/cpal?branch=osx-screencapturekit#96d4da121b7d949677ac5b6887413a9185fd7f39"
 dependencies = [
  "alsa",
  "cidre",

src/client.rs

@@ -95,7 +95,10 @@ pub use super::lang::*;
 
 #[cfg(not(target_os = "linux"))]
 mod audio_playback;
+#[cfg(target_os = "windows")]
+mod audio_playback_recovery;
 #[cfg(all(test, not(target_os = "linux")))]
+#[path = "client/tests/audio_state_tests.rs"]
 mod audio_state_tests;
 pub mod file_trait;
 pub mod helper;
@@ -2083,6 +2086,8 @@ pub struct AudioHandler {
     device_channel: u16,
     #[cfg(not(target_os = "linux"))]
     playback_status: Arc<audio_playback::AudioPlaybackStatus>,
+    #[cfg(target_os = "windows")]
+    playback_recovery: audio_playback_recovery::PlaybackRecovery,
 }
 
 #[cfg(not(target_os = "linux"))]
@@ -2390,22 +2395,53 @@ impl AudioHandler {
 
     /// Handle audio format and create an audio decoder.
     pub fn handle_format(&mut self, f: AudioFormat) {
+        self.handle_format_with_start(f, Self::start_audio);
+    }
+
+    fn handle_format_with_start(
+        &mut self,
+        f: AudioFormat,
+        start: impl FnOnce(&mut Self, AudioFormat) -> ResultType<()>,
+    ) {
         if !is_supported_audio_channel_count(f.channels) {
             log::error!("Unsupported audio channel count: {}", f.channels);
             return;
         }
         match AudioDecoder::new(f.sample_rate, if f.channels > 1 { Stereo } else { Mono }) {
             Ok(d) => {
+                #[cfg(target_os = "windows")]
+                let playback_failed = self.cancel_pending_playback();
                 #[cfg(target_os = "linux")]
                 let keep_existing_stream = self.simple.is_some()
                     && self.sample_rate.0 == f.sample_rate
                     && u32::from(self.channels) == f.channels;
                 #[cfg(not(target_os = "linux"))]
-                let keep_existing_stream = false;
+                let keep_existing_stream = self.audio_stream.is_some()
+                    && self.sample_rate.0 == f.sample_rate
+                    && u32::from(self.channels) == f.channels;
                 let buffer = vec![0.; f.sample_rate as usize * f.channels as usize];
+                #[cfg(not(target_os = "linux"))]
+                let mut previous = std::mem::take(self);
+                #[cfg(target_os = "windows")]
+                self.prepare_playback(&f);
                 self.audio_decoder = Some((d, buffer));
                 self.channels = f.channels as _;
-                let result = self.start_audio(f);
+                let result = start(self, f);
+                #[cfg(target_os = "windows")]
+                let keep_existing_stream = keep_existing_stream
+                    && !playback_failed
+                    && !previous.playback_recovery.report_pending();
+                #[cfg(not(target_os = "linux"))]
+                if result.is_err() && keep_existing_stream {
+                    // The restarted capture has new Opus history even when output startup fails.
+                    previous.audio_decoder = self.audio_decoder.take();
+                    *self = previous;
+                    self.handle_audio_start_result(result, true);
+                    return;
+                }
+                #[cfg(target_os = "windows")]
+                self.finish_playback_replacement(result, keep_existing_stream.then_some(previous));
+                #[cfg(not(target_os = "windows"))]
                 self.handle_audio_start_result(result, keep_existing_stream);
             }
             Err(err) => {
@@ -2493,6 +2529,9 @@ impl AudioHandler {
         device: &Device,
     ) -> ResultType<()> {
         self.device_channel = config.channels;
+        #[cfg(target_os = "windows")]
+        let err_fn = self.playback_recovery.new_error_callback();
+        #[cfg(not(target_os = "windows"))]
         let err_fn = move |err| {
             // too many errors, will improve later
             log::trace!("an error occurred on stream: {}", err);
@@ -4065,7 +4104,11 @@ pub fn start_audio_thread() -> MediaSender {
     std::thread::spawn(move || {
         let mut audio_handler = AudioHandler::default();
         loop {
-            if let Ok(data) = audio_receiver.recv() {
+            #[cfg(target_os = "windows")]
+            let received = audio_handler.receive_audio(&audio_receiver);
+            #[cfg(not(target_os = "windows"))]
+            let received = audio_receiver.recv();
+            if let Ok(data) = received {
                 match data {
                     MediaData::AudioFrame(af) => {
                         audio_handler.handle_frame(*af);

src/client/audio_playback_recovery.rs

@@ -0,0 +1,184 @@
+use super::{AudioDecoder, AudioFormat, AudioHandler, MediaData, Mono, Stereo};
+use cpal::StreamError;
+use crossbeam_queue::SegQueue;
+use hbb_common::{log, tokio::time::Instant, ResultType};
+use std::{
+    sync::{atomic::Ordering, mpsc, Arc},
+    time::Duration,
+};
+
+const RECOVERY_INTERVAL: Duration = Duration::from_secs(1);
+pub(super) const STARTUP_CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(5);
+// The pinned WASAPI backend reports this warning but keeps its worker running.
+const PRIORITY_WARNING_PREFIX: &str = "SetThreadPriority failed: ";
+
+#[path = "audio_playback_startup.rs"]
+mod startup;
+
+#[derive(Default)]
+pub(super) struct PlaybackRecovery {
+    pub(super) errors: Arc<SegQueue<StreamError>>,
+    format: Option<AudioFormat>,
+    pub(super) retry_at: Option<Instant>,
+    restart_not_before: Option<Instant>,
+    awaiting_callback: bool,
+    startup_deadline: Option<Instant>,
+    pending_output: Option<Box<AudioHandler>>,
+}
+
+impl PlaybackRecovery {
+    pub(super) fn new_error_callback(&mut self) -> impl FnMut(StreamError) + Send + 'static {
+        self.errors = Default::default();
+        let errors = self.errors.clone();
+        move |error| errors.push(error)
+    }
+
+    pub(super) fn report_pending(&self) -> bool {
+        let mut failed = false;
+        while let Some(error) = self.errors.pop() {
+            if matches!(&error, StreamError::BackendSpecific { err }
+                if err.description.starts_with(PRIORITY_WARNING_PREFIX))
+            {
+                log::warn!("Audio playback nonterminal priority warning: {error}");
+            } else {
+                log::error!("Audio playback stream failed: {error}");
+                failed = true;
+            }
+        }
+        failed
+    }
+}
+
+impl AudioHandler {
+    fn clear_playback_stream(&mut self) {
+        // Dropping CPAL may join its worker; run this on the owner, not its callback.
+        self.audio_stream = None;
+        self.playback_recovery.report_pending();
+        self.playback_status.report_errors();
+        let recovery = std::mem::take(self).playback_recovery;
+        self.playback_recovery.format = recovery.format;
+        self.playback_recovery.retry_at = recovery.retry_at;
+        self.playback_recovery.restart_not_before = recovery.restart_not_before;
+    }
+
+    pub(super) fn prepare_playback(&mut self, format: &AudioFormat) {
+        self.clear_playback_stream();
+        self.playback_recovery.format = Some(format.clone());
+        self.playback_recovery.retry_at = None;
+        self.playback_recovery.restart_not_before = None;
+    }
+
+    pub(super) fn finish_playback_start(&mut self, result: ResultType<()>) {
+        let now = Instant::now();
+        let retry_at = now + RECOVERY_INTERVAL;
+        self.playback_recovery.restart_not_before = Some(retry_at);
+        match result {
+            Ok(()) => {
+                self.playback_recovery.retry_at = None;
+                self.playback_recovery.awaiting_callback = true;
+                self.playback_recovery.startup_deadline = Some(now + STARTUP_CONFIRMATION_TIMEOUT);
+                log::info!("Audio playback stream opened; waiting for output callback");
+            }
+            Err(error) => {
+                self.clear_playback_stream();
+                self.playback_recovery.retry_at = Some(retry_at);
+                log::error!(
+                    "Audio playback start failed: {error:#}; retrying in {RECOVERY_INTERVAL:?}"
+                );
+            }
+        }
+    }
+
+    fn playback_start_timed_out(&mut self, now: Instant) -> bool {
+        if !self.playback_recovery.awaiting_callback
+            || self.playback_status.ready.load(Ordering::Acquire)
+            || !self
+                .playback_recovery
+                .startup_deadline
+                .is_some_and(|due| now >= due)
+        {
+            return false;
+        }
+        self.playback_recovery.awaiting_callback = false;
+        self.playback_recovery.startup_deadline = None;
+        log::error!("Audio playback start timed out waiting for output callback");
+        true
+    }
+
+    fn restart_playback(&mut self, format: AudioFormat) -> ResultType<()> {
+        let channels = if format.channels > 1 { Stereo } else { Mono };
+        let decoder = AudioDecoder::new(format.sample_rate, channels)?;
+        let buffer = vec![0.; format.sample_rate as usize * format.channels as usize];
+        let channel_count = format.channels as _;
+        self.start_audio(format)?;
+        self.channels = channel_count;
+        self.audio_decoder = Some((decoder, buffer));
+        Ok(())
+    }
+
+    pub(super) fn recover_playback_with(
+        &mut self,
+        now: Instant,
+        restart: impl FnOnce(&mut Self, AudioFormat) -> ResultType<()>,
+    ) {
+        let failed = self.resolve_pending_playback(now).unwrap_or_else(|| {
+            self.playback_recovery.report_pending() || self.playback_start_timed_out(now)
+        });
+        if failed {
+            self.clear_playback_stream();
+            self.playback_recovery.retry_at = Some(
+                self.playback_recovery
+                    .restart_not_before
+                    .map_or(now, |due| due.max(now)),
+            );
+        }
+        if self.playback_recovery.awaiting_callback
+            && self.playback_status.ready.load(Ordering::Acquire)
+        {
+            self.playback_recovery.awaiting_callback = false;
+            self.playback_recovery.startup_deadline = None;
+            log::info!("Audio playback output callback started");
+        }
+        if !self
+            .playback_recovery
+            .retry_at
+            .is_some_and(|due| now >= due)
+        {
+            return;
+        }
+        let Some(format) = self.playback_recovery.format.clone() else {
+            return;
+        };
+        log::info!("Recreating audio playback on the current default output device");
+        let result = restart(self, format);
+        self.finish_playback_start(result);
+    }
+
+    pub(super) fn receive_audio(
+        &mut self,
+        receiver: &mpsc::Receiver<MediaData>,
+    ) -> Result<MediaData, mpsc::RecvError> {
+        receive_with_recovery(receiver, RECOVERY_INTERVAL, || {
+            self.recover_playback_with(Instant::now(), Self::restart_playback);
+        })
+    }
+}
+
+pub(super) fn receive_with_recovery(
+    receiver: &mpsc::Receiver<MediaData>,
+    interval: Duration,
+    mut recover: impl FnMut(),
+) -> Result<MediaData, mpsc::RecvError> {
+    loop {
+        match receiver.recv_timeout(interval) {
+            Ok(data) => {
+                if !matches!(data, MediaData::AudioFormat(_)) {
+                    recover();
+                }
+                return Ok(data);
+            }
+            Err(mpsc::RecvTimeoutError::Timeout) => recover(),
+            Err(mpsc::RecvTimeoutError::Disconnected) => return Err(mpsc::RecvError),
+        }
+    }
+}

src/client/audio_playback_startup.rs

@@ -0,0 +1,63 @@
+use super::{AudioHandler, Instant, Ordering, ResultType};
+use hbb_common::log;
+
+impl AudioHandler {
+    pub(in crate::client) fn cancel_pending_playback(&mut self) -> bool {
+        // Format messages bypass recovery; retain a usable candidate before superseding it.
+        let failed = self
+            .resolve_pending_playback(Instant::now())
+            .unwrap_or(false);
+        if let Some(mut pending) = self.playback_recovery.pending_output.take() {
+            pending.audio_stream = None;
+            pending.playback_recovery.report_pending();
+            pending.playback_status.report_errors();
+        }
+        failed
+    }
+
+    pub(in crate::client) fn finish_playback_replacement(
+        &mut self,
+        result: ResultType<()>,
+        previous: Option<Self>,
+    ) {
+        self.finish_playback_start(result);
+        let Some(mut previous) = previous else {
+            return;
+        };
+        previous.audio_decoder = self.audio_decoder.take();
+        let candidate = std::mem::replace(self, previous);
+        self.playback_recovery.pending_output = Some(Box::new(candidate));
+        log::info!("Audio playback replacement pending; continuing on the compatible output");
+        self.recover_playback_with(Instant::now(), Self::restart_playback);
+    }
+
+    pub(super) fn resolve_pending_playback(&mut self, now: Instant) -> Option<bool> {
+        let mut candidate = self.playback_recovery.pending_output.take()?;
+        let candidate_failed =
+            candidate.playback_recovery.report_pending() || candidate.playback_start_timed_out(now);
+        let previous_failed =
+            self.playback_recovery.report_pending() || self.playback_start_timed_out(now);
+        if candidate_failed {
+            self.playback_recovery.restart_not_before =
+                candidate.playback_recovery.restart_not_before;
+            candidate.audio_stream = None;
+            candidate.playback_recovery.report_pending();
+            candidate.playback_status.report_errors();
+            if !previous_failed {
+                log::error!("Audio playback replacement failed before startup confirmation; keeping the existing compatible stream");
+            }
+            return Some(previous_failed);
+        }
+        if candidate.playback_status.ready.load(Ordering::Acquire) || previous_failed {
+            candidate.audio_decoder = self.audio_decoder.take();
+            self.audio_stream = None;
+            self.playback_recovery.report_pending();
+            self.playback_status.report_errors();
+            *self = *candidate;
+            return Some(false);
+        }
+        self.playback_recovery.pending_output = Some(candidate);
+        // A second active-queue read could discard a healthy pending candidate.
+        Some(false)
+    }
+}

src/client/audio_state_tests.rs

@@ -1,113 +0,0 @@
-use super::{create_audio_resampler, AudioDecoder, AudioFrame, AudioHandler, Stereo};
-use cpal::traits::StreamTrait;
-use hbb_common::anyhow::anyhow;
-use magnum_opus::{Application::LowDelay, Encoder};
-use ringbuf::{ring_buffer::RbBase, Rb};
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Arc,
-};
-
-const INPUT_RATE: u32 = 24_000;
-const OUTPUT_RATE: u32 = 48_000;
-const CHANNELS: u16 = 2;
-const PACKETS_PER_SECOND: usize = 100;
-const MAX_PACKET_BYTES: usize = 4_096;
-const SAMPLE_VALUE: f32 = 0.25;
-
-struct TrackedAudioStream(Arc<AtomicBool>);
-
-impl StreamTrait for TrackedAudioStream {
-    fn play(&self) -> Result<(), cpal::PlayStreamError> {
-        Ok(())
-    }
-
-    fn pause(&self) -> Result<(), cpal::PauseStreamError> {
-        Ok(())
-    }
-}
-
-impl Drop for TrackedAudioStream {
-    fn drop(&mut self) {
-        self.0.store(true, Ordering::SeqCst);
-    }
-}
-
-fn decoder(sample_rate: u32) -> (AudioDecoder, Vec<f32>) {
-    (
-        AudioDecoder::new(sample_rate, Stereo).unwrap(),
-        vec![0.0; sample_rate as usize * CHANNELS as usize],
-    )
-}
-
-fn active_handler(input_rate: u32) -> (AudioHandler, Arc<AtomicBool>) {
-    let dropped = Arc::new(AtomicBool::new(false));
-    let handler = AudioHandler {
-        audio_decoder: Some(decoder(input_rate)),
-        audio_resampler: create_audio_resampler(input_rate, OUTPUT_RATE, CHANNELS).unwrap(),
-        sample_rate: (input_rate, OUTPUT_RATE),
-        audio_stream: Some(Box::new(TrackedAudioStream(dropped.clone()))),
-        channels: CHANNELS,
-        device_channel: CHANNELS,
-        ..Default::default()
-    };
-    handler.playback_status.ready.store(true, Ordering::Release);
-    (handler, dropped)
-}
-
-fn audio_frame() -> AudioFrame {
-    let samples = OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize;
-    let mut encoder = Encoder::new(OUTPUT_RATE, Stereo, LowDelay).unwrap();
-    AudioFrame {
-        data: encoder
-            .encode_vec_float(&vec![SAMPLE_VALUE; samples], MAX_PACKET_BYTES)
-            .unwrap()
-            .into(),
-        ..Default::default()
-    }
-}
-
-#[test]
-fn failed_format_change_discards_old_playback_state() {
-    let (mut handler, dropped) = active_handler(INPUT_RATE);
-    handler
-        .audio_buffer
-        .0
-        .lock()
-        .unwrap()
-        .push_slice(&[SAMPLE_VALUE; CHANNELS as usize]);
-    handler.audio_decoder = Some(decoder(OUTPUT_RATE));
-    handler.sample_rate = (OUTPUT_RATE, OUTPUT_RATE);
-
-    handler.handle_audio_start_result(
-        Err(anyhow!("Injected output stream startup failure")),
-        false,
-    );
-
-    assert!(dropped.load(Ordering::SeqCst));
-    assert!(handler.audio_stream.is_none());
-    assert!(handler.audio_resampler.is_none());
-    assert!(handler.audio_decoder.is_none());
-    assert!(!handler.playback_status.ready.load(Ordering::Acquire));
-    handler.handle_frame(audio_frame());
-    assert_eq!(handler.audio_buffer.0.lock().unwrap().occupied_len(), 0);
-}
-
-#[test]
-fn successful_start_or_compatible_failure_preserves_audio_packet_duration() {
-    for result in [
-        Ok(()),
-        Err(anyhow!("Injected compatible stream replacement failure")),
-    ] {
-        let (mut handler, dropped) = active_handler(OUTPUT_RATE);
-
-        handler.handle_audio_start_result(result, true);
-        handler.handle_frame(audio_frame());
-
-        assert!(!dropped.load(Ordering::SeqCst));
-        assert_eq!(
-            handler.audio_buffer.0.lock().unwrap().occupied_len(),
-            OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize
-        );
-    }
-}

src/client/tests/audio_playback_recovery_tests.rs

@@ -0,0 +1,226 @@
+use super::*;
+use crate::client::{
+    audio_playback::AudioPlaybackStatus, audio_playback_recovery::STARTUP_CONFIRMATION_TIMEOUT,
+};
+use cpal::StreamError;
+use crossbeam_queue::SegQueue;
+use hbb_common::tokio::time::Instant;
+use std::time::Duration;
+
+const AFTER_COOLDOWN: Duration = Duration::from_secs(2);
+
+type PendingOutput = (
+    Arc<AtomicBool>,
+    Arc<AudioPlaybackStatus>,
+    Arc<SegQueue<StreamError>>,
+);
+
+fn install_output(handler: &mut AudioHandler, dropped: Arc<AtomicBool>) {
+    handler.sample_rate = (INPUT_RATE, OUTPUT_RATE);
+    handler.device_channel = CHANNELS;
+    handler.audio_stream = Some(Box::new(TrackedAudioStream(dropped)));
+    handler.playback_status.ready.store(true, Ordering::Release);
+}
+
+fn recovery_handler() -> (AudioHandler, Arc<AtomicBool>) {
+    let dropped = Arc::new(AtomicBool::new(false));
+    let mut handler = AudioHandler::default();
+    handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), |candidate, _| {
+        install_output(candidate, dropped.clone());
+        Ok(())
+    });
+    (handler, dropped)
+}
+
+fn begin_pending(handler: &mut AudioHandler) -> PendingOutput {
+    let dropped = Arc::new(AtomicBool::new(false));
+    let mut state = None;
+    handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), |candidate, _| {
+        install_output(candidate, dropped.clone());
+        candidate
+            .playback_status
+            .ready
+            .store(false, Ordering::Release);
+        state = Some((
+            candidate.playback_status.clone(),
+            candidate.playback_recovery.errors.clone(),
+        ));
+        Ok(())
+    });
+    let (status, errors) = state.unwrap();
+    (dropped, status, errors)
+}
+
+#[test]
+fn unconfirmed_start_retries_without_callback_or_error() {
+    let dropped = Arc::new(AtomicBool::new(false));
+    let mut handler = AudioHandler::default();
+    handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), |candidate, _| {
+        install_output(candidate, dropped.clone());
+        candidate
+            .playback_status
+            .ready
+            .store(false, Ordering::Release);
+        Ok(())
+    });
+    handler.recover_playback_with(Instant::now(), |_, _| {
+        panic!("Startup confirmation deadline has not elapsed")
+    });
+    assert!(!dropped.load(Ordering::SeqCst));
+    let expired = Instant::now() + STARTUP_CONFIRMATION_TIMEOUT;
+    let mut attempts = 0;
+    handler.recover_playback_with(expired, |candidate, requested| {
+        attempts += 1;
+        assert!(dropped.load(Ordering::SeqCst));
+        assert_eq!(requested, format(INPUT_RATE, CHANNELS));
+        install_output(candidate, Arc::new(AtomicBool::new(false)));
+        Ok(())
+    });
+    assert_eq!(attempts, 1);
+    handler.recover_playback_with(expired + STARTUP_CONFIRMATION_TIMEOUT, |_, _| {
+        panic!("Confirmed output must not be reopened")
+    });
+}
+
+#[test]
+fn already_terminal_candidate_cannot_replace_compatible_output() {
+    let (mut handler, dropped) = recovery_handler();
+    let candidate_dropped = Arc::new(AtomicBool::new(false));
+    let old_buffer = handler.audio_buffer.0.clone();
+    handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), |candidate, _| {
+        install_output(candidate, candidate_dropped.clone());
+        candidate
+            .playback_recovery
+            .errors
+            .push(StreamError::DeviceNotAvailable);
+        Ok(())
+    });
+    assert!(!dropped.load(Ordering::SeqCst));
+    assert!(candidate_dropped.load(Ordering::SeqCst));
+    assert!(Arc::ptr_eq(&old_buffer, &handler.audio_buffer.0));
+    handler.recover_playback_with(Instant::now() + AFTER_COOLDOWN, |_, _| {
+        panic!("Compatible active output must not be reopened")
+    });
+}
+
+#[test]
+fn pending_output_keeps_playing_and_transfers_decoder_history_on_commit() {
+    let (mut handler, old_dropped) = recovery_handler();
+    let old_buffer = handler.audio_buffer.0.clone();
+    let (_, status, _) = begin_pending(&mut handler);
+    let frame = audio_frame();
+    let (mut reference, mut expected) = decoder(INPUT_RATE);
+    reference
+        .decode_float(&frame.data, &mut expected, false)
+        .unwrap();
+    handler.handle_frame(frame.clone());
+    assert!(Arc::ptr_eq(&old_buffer, &handler.audio_buffer.0));
+    assert!(!drain_audio(&handler).is_empty());
+    status.ready.store(true, Ordering::Release);
+    handler.recover_playback_with(Instant::now(), |_, _| panic!("Candidate already exists"));
+    assert!(old_dropped.load(Ordering::SeqCst));
+    assert!(Arc::ptr_eq(&status, &handler.playback_status));
+    let samples = reference
+        .decode_float(&frame.data, &mut expected, false)
+        .unwrap()
+        * CHANNELS as usize;
+    handler.handle_frame(frame);
+    assert_eq!(
+        &handler.audio_decoder.as_ref().unwrap().1[..samples],
+        &expected[..samples]
+    );
+}
+
+#[test]
+fn rollback_keeps_the_restarted_decoders_accumulated_history() {
+    let (mut handler, old_dropped) = recovery_handler();
+    let frame = audio_frame();
+    handler.handle_frame(frame.clone());
+    let (candidate_dropped, _, errors) = begin_pending(&mut handler);
+    let (mut reference, mut expected) = decoder(INPUT_RATE);
+    handler.handle_frame(frame.clone());
+    reference
+        .decode_float(&frame.data, &mut expected, false)
+        .unwrap();
+    errors.push(StreamError::DeviceNotAvailable);
+    handler.recover_playback_with(Instant::now(), |_, _| panic!("Old output still works"));
+    let samples = reference
+        .decode_float(&frame.data, &mut expected, false)
+        .unwrap()
+        * CHANNELS as usize;
+    handler.handle_frame(frame);
+    assert!(!old_dropped.load(Ordering::SeqCst));
+    assert!(candidate_dropped.load(Ordering::SeqCst));
+    assert_eq!(
+        &handler.audio_decoder.as_ref().unwrap().1[..samples],
+        &expected[..samples]
+    );
+}
+
+#[test]
+fn later_compatible_format_retires_only_the_pending_attempt() {
+    let (mut handler, old_dropped) = recovery_handler();
+    let (first_dropped, _, first_errors) = begin_pending(&mut handler);
+    let (second_dropped, second_status, _) = begin_pending(&mut handler);
+    assert!(first_dropped.load(Ordering::SeqCst));
+    assert!(!old_dropped.load(Ordering::SeqCst));
+    first_errors.push(StreamError::DeviceNotAvailable);
+    second_status.ready.store(true, Ordering::Release);
+    handler.recover_playback_with(Instant::now(), |_, _| {
+        panic!("Retired attempt affected current output")
+    });
+    assert!(old_dropped.load(Ordering::SeqCst));
+    assert!(!second_dropped.load(Ordering::SeqCst));
+    assert!(Arc::ptr_eq(&second_status, &handler.playback_status));
+}
+
+#[test]
+fn both_outputs_failing_retains_format_and_paces_recovery() {
+    let (mut handler, old_dropped) = recovery_handler();
+    let old_errors = handler.playback_recovery.errors.clone();
+    let (candidate_dropped, _, errors) = begin_pending(&mut handler);
+    old_errors.push(StreamError::DeviceNotAvailable);
+    errors.push(StreamError::DeviceNotAvailable);
+    handler.recover_playback_with(Instant::now(), |_, _| panic!("Retry must be paced"));
+    assert!(old_dropped.load(Ordering::SeqCst));
+    assert!(candidate_dropped.load(Ordering::SeqCst));
+    assert!(handler.audio_stream.is_none());
+    let due = handler.playback_recovery.retry_at.unwrap();
+    let mut attempts = 0;
+    handler.recover_playback_with(due, |_, requested| {
+        attempts += 1;
+        assert_eq!(requested, format(INPUT_RATE, CHANNELS));
+        Ok(())
+    });
+    assert_eq!(attempts, 1);
+}
+
+#[test]
+fn superseding_format_keeps_ready_candidate_when_active_output_failed() {
+    let (mut handler, old_dropped) = recovery_handler();
+    let old_errors = handler.playback_recovery.errors.clone();
+    let (candidate_dropped, candidate_status, _) = begin_pending(&mut handler);
+    candidate_status.ready.store(true, Ordering::Release);
+    old_errors.push(StreamError::DeviceNotAvailable);
+    handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), failed_output);
+    assert!(!candidate_dropped.load(Ordering::SeqCst));
+    assert!(old_dropped.load(Ordering::SeqCst));
+    assert!(Arc::ptr_eq(&candidate_status, &handler.playback_status));
+    assert!(handler.audio_decoder.is_some());
+    assert!(handler.playback_recovery.retry_at.is_none());
+}
+
+#[test]
+fn superseding_format_preserves_failure_when_both_outputs_failed() {
+    let (mut handler, old_dropped) = recovery_handler();
+    let old_errors = handler.playback_recovery.errors.clone();
+    let (candidate_dropped, status, errors) = begin_pending(&mut handler);
+    status.ready.store(true, Ordering::Release);
+    old_errors.push(StreamError::DeviceNotAvailable);
+    errors.push(StreamError::DeviceNotAvailable);
+    handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), failed_output);
+    assert!(candidate_dropped.load(Ordering::SeqCst));
+    assert!(old_dropped.load(Ordering::SeqCst));
+    assert!(handler.audio_stream.is_none());
+    assert!(handler.playback_recovery.retry_at.is_some());
+}

src/client/tests/audio_state_tests.rs

@@ -0,0 +1,128 @@
+use super::{create_audio_resampler, AudioDecoder, AudioFormat, AudioFrame, AudioHandler, Stereo};
+use cpal::traits::StreamTrait;
+use hbb_common::{anyhow::anyhow, ResultType};
+use magnum_opus::{Application::LowDelay, Encoder};
+use ringbuf::Rb;
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
+
+const INPUT_RATE: u32 = 24_000;
+const OUTPUT_RATE: u32 = 48_000;
+const CHANNELS: u16 = 2;
+const PACKETS_PER_SECOND: usize = 100;
+const MAX_PACKET_BYTES: usize = 4_096;
+const SAMPLE_VALUE: f32 = 0.25;
+const MONO_CHANNELS: u16 = 1;
+
+#[cfg(target_os = "windows")]
+#[path = "audio_playback_recovery_tests.rs"]
+mod recovery_tests;
+
+struct TrackedAudioStream(Arc<AtomicBool>);
+
+impl StreamTrait for TrackedAudioStream {
+    fn play(&self) -> Result<(), cpal::PlayStreamError> {
+        Ok(())
+    }
+
+    fn pause(&self) -> Result<(), cpal::PauseStreamError> {
+        Ok(())
+    }
+}
+
+impl Drop for TrackedAudioStream {
+    fn drop(&mut self) {
+        self.0.store(true, Ordering::SeqCst);
+    }
+}
+
+fn decoder(sample_rate: u32) -> (AudioDecoder, Vec<f32>) {
+    (
+        AudioDecoder::new(sample_rate, Stereo).unwrap(),
+        vec![0.0; sample_rate as usize * CHANNELS as usize],
+    )
+}
+
+fn active_handler(input_rate: u32) -> (AudioHandler, Arc<AtomicBool>) {
+    let dropped = Arc::new(AtomicBool::new(false));
+    let handler = AudioHandler {
+        audio_decoder: Some(decoder(input_rate)),
+        audio_resampler: create_audio_resampler(input_rate, OUTPUT_RATE, CHANNELS).unwrap(),
+        sample_rate: (input_rate, OUTPUT_RATE),
+        audio_stream: Some(Box::new(TrackedAudioStream(dropped.clone()))),
+        channels: CHANNELS,
+        device_channel: CHANNELS,
+        ..Default::default()
+    };
+    handler.playback_status.ready.store(true, Ordering::Release);
+    (handler, dropped)
+}
+
+fn audio_frame() -> AudioFrame {
+    let samples = OUTPUT_RATE as usize / PACKETS_PER_SECOND * CHANNELS as usize;
+    let mut encoder = Encoder::new(OUTPUT_RATE, Stereo, LowDelay).unwrap();
+    AudioFrame {
+        data: encoder
+            .encode_vec_float(&vec![SAMPLE_VALUE; samples], MAX_PACKET_BYTES)
+            .unwrap()
+            .into(),
+        ..Default::default()
+    }
+}
+
+fn failed_output(candidate: &mut AudioHandler, _: AudioFormat) -> ResultType<()> {
+    candidate.sample_rate = (INPUT_RATE, INPUT_RATE);
+    candidate.device_channel = MONO_CHANNELS;
+    candidate
+        .audio_buffer
+        .resize(INPUT_RATE as _, MONO_CHANNELS as _);
+    candidate
+        .playback_status
+        .ready
+        .store(false, Ordering::Release);
+    Err(anyhow!("Injected playback failure"))
+}
+
+fn format(sample_rate: u32, channels: u16) -> AudioFormat {
+    AudioFormat {
+        sample_rate,
+        channels: u32::from(channels),
+        ..Default::default()
+    }
+}
+
+#[test]
+fn identical_format_failure_preserves_playback_and_resampler_history() {
+    let (mut handler, dropped) = active_handler(INPUT_RATE);
+    let (mut reference, _) = active_handler(INPUT_RATE);
+    let (mut retained_decoder, _) = active_handler(INPUT_RATE);
+    retained_decoder.handle_frame(audio_frame());
+    retained_decoder.handle_frame(audio_frame());
+    handler.handle_frame(audio_frame());
+    reference.handle_frame(audio_frame());
+    let buffer = handler.audio_buffer.0.clone();
+    let status = handler.playback_status.clone();
+
+    handler.handle_format_with_start(format(INPUT_RATE, CHANNELS), failed_output);
+    reference.audio_decoder = Some(decoder(INPUT_RATE));
+    handler.handle_frame(audio_frame());
+    reference.handle_frame(audio_frame());
+
+    assert!(!dropped.load(Ordering::SeqCst));
+    assert!(Arc::ptr_eq(&buffer, &handler.audio_buffer.0));
+    assert!(Arc::ptr_eq(&status, &handler.playback_status));
+    assert_eq!(handler.sample_rate, (INPUT_RATE, OUTPUT_RATE));
+    assert_eq!(handler.device_channel, CHANNELS);
+    assert!(handler.playback_status.ready.load(Ordering::Acquire));
+    let expected = drain_audio(&reference);
+    let actual = drain_audio(&handler);
+    assert!(!actual.is_empty());
+    assert_ne!(drain_audio(&retained_decoder), expected);
+    assert_eq!(actual, expected);
+}
+
+fn drain_audio(handler: &AudioHandler) -> Vec<f32> {
+    handler.audio_buffer.0.lock().unwrap().pop_iter().collect()
+}

src/server/audio_service.rs

@@ -391,6 +391,15 @@ mod cpal_impl {
         if !audio_input.is_empty() {
             return get_audio_input(&audio_input);
         }
+        // The pinned CPAL uses event-driven WASAPI loopback here. Windows versions
+        // before Windows 10 1703 do not signal capture events, so system audio does
+        // not work on Win7. #16095 kept the same CPAL revision and loopback path;
+        // this limitation predates that PR.
+        // Ordinary microphone input is supported on Win7 and uses the branch above.
+        // #16095 added its callback-to-encoder wake dependency; see CapturePcmSender::wake
+        // for the new scheduling risk, whose audible impact on Win7 is unmeasured.
+        // https://learn.microsoft.com/en-us/windows/win32/coreaudio/loopback-recording
+        // https://learn.microsoft.com/en-us/windows/win32/coreaudio/capturesharedeventdriven
         let device = HOST
             .default_output_device()
             .with_context(|| "Failed to get default output device for loopback")?;

src/server/audio_service/audio_capture_queue.rs

@@ -97,6 +97,7 @@ impl Drop for CaptureEncoderWorker {
     fn drop(&mut self) {
         self.stop.store(true, Ordering::Release);
         if let Some(handle) = self.handle.take() {
+            // Owner-thread shutdown already waits via join(); see CapturePcmSender::wake for Win7.
             handle.thread().unpark();
             if let Err(error) = handle.join() {
                 log::error!("Failed to join audio encoder thread: {error:?}");
@@ -209,6 +210,16 @@ impl CapturePcmSender {
 
     fn wake(&self) {
         if let Some(thread) = self.handoff.wake_thread.get() {
+            // #16095 moved Opus encoding and message submission from the capture callback to a worker.
+            // Previously, the callback did that work directly, with allocations and blocking locks.
+            // On Win7 with Rust 1.75, if the worker is descheduled after publishing PARKED but before
+            // NtWaitForKeyedEvent, unpark() waits in NtReleaseKeyedEvent until the worker enters that wait.
+            // It does not wait for encoding; park_timeout() does not bound the callback's wait.
+            // Delays can cause gaps or stall teardown; a Win7 microphone regression has not been measured.
+            // System loopback already failed on Win7 before #16095 (see cpal_impl::get_device),
+            // so the affected path is microphone/input-device capture, including outgoing voice calls.
+            // Accept this risk to preserve Win7 input capture without a separate legacy notifier.
+            // https://github.com/rust-lang/rust/blob/1.75.0/library/std/src/sys/windows/thread_parking.rs
             thread.unpark();
         }
     }