fix(coreaudio): honour the full timeout when waiting for a sample rate change

CHANGELOG.md

@@ -51,6 +51,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **AudioWorklet**: Fix dropouts in output streams when the callback buffer grows.
 - **CoreAudio**: Fix the device running at a different sample rate from the stream on hardware that reports a continuous rate range.
 - **CoreAudio**: Fix `supported_configs()` only reporting `F32`, even on hardware that also supports other sample formats.
+- **CoreAudio**: Fix sample rate changes timing out early when the device reports other rates first.
 - **JACK**: Channel enumeration is capped at the physical system port count again.
 - **JACK**: Streams no longer panic when the server delivers a larger period than the negotiated buffer size.
 - **PipeWire**: Fix an empty chunk being emitted when a cycle requests no frames.

src/host/coreaudio/macos/device.rs

@@ -5,7 +5,7 @@ use std::{
     sync::{
         Arc, Mutex,
         atomic::{AtomicBool, AtomicUsize, Ordering},
-        mpsc::{RecvTimeoutError, channel},
+        mpsc::{Receiver, RecvTimeoutError, channel},
     },
     time::{Duration, Instant},
 };
@@ -191,37 +191,57 @@ fn set_sample_rate(
         // Wait for the reported_rate to change.
         //
         // This should not take longer than a few ms. Use the caller's timeout if provided,
-        // otherwise default to 1 second. We loop over potentially several events from the
-        // channel to ensure that we catch the expected change in sample rate.
-        let mut remaining = timeout.unwrap_or(Duration::from_secs(1));
-        let start = Instant::now();
-        loop {
-            match receiver.recv_timeout(remaining) {
-                Ok(reported_rate) => {
-                    if (reported_rate - target_sample_rate as f64).abs() < 1.0 {
-                        break;
-                    }
-                }
-                Err(RecvTimeoutError::Timeout) => {
-                    return Err(Error::with_message(
-                        ErrorKind::DeviceNotAvailable,
-                        "Sample rate update timed out",
-                    ));
-                }
-                Err(RecvTimeoutError::Disconnected) => {
-                    return Err(Error::with_message(
-                        ErrorKind::StreamInvalidated,
-                        "Sample rate listener disconnected unexpectedly",
-                    ));
+        // otherwise default to 1 second.
+        wait_for_rate(
+            &receiver,
+            target_sample_rate,
+            timeout.unwrap_or(Duration::from_secs(1)),
+        )?;
+        // listener dropped here; its Drop impl calls unregister() automatically.
+    }
+    Ok(())
+}
+
+/// Block until the rate listener reports `target_sample_rate`, giving up after `timeout`.
+///
+/// Notifications carrying some other rate can arrive first, so `timeout` bounds the whole wait
+/// rather than each individual receive.
+fn wait_for_rate(
+    receiver: &Receiver<f64>,
+    target_sample_rate: SampleRate,
+    timeout: Duration,
+) -> Result<(), Error> {
+    let deadline = Instant::now() + timeout;
+
+    loop {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        if remaining.is_zero() {
+            return Err(Error::with_message(
+                ErrorKind::DeviceNotAvailable,
+                "Sample rate update timed out",
+            ));
+        }
+
+        match receiver.recv_timeout(remaining) {
+            Ok(reported_rate) => {
+                if (reported_rate - target_sample_rate as f64).abs() < 1.0 {
+                    return Ok(());
                 }
             }
-            remaining = remaining
-                .checked_sub(start.elapsed())
-                .unwrap_or(Duration::ZERO);
+            Err(RecvTimeoutError::Timeout) => {
+                return Err(Error::with_message(
+                    ErrorKind::DeviceNotAvailable,
+                    "Sample rate update timed out",
+                ));
+            }
+            Err(RecvTimeoutError::Disconnected) => {
+                return Err(Error::with_message(
+                    ErrorKind::StreamInvalidated,
+                    "Sample rate listener disconnected unexpectedly",
+                ));
+            }
         }
-        // listener dropped here; its Drop impl calls unregister() automatically.
     }
-    Ok(())
 }
 
 #[derive(Clone, Copy)]
@@ -1202,3 +1222,46 @@ pub(crate) fn get_device_buffer_frame_size(
     )?;
     Ok(frames as usize)
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::mpsc::channel;
+    use std::time::{Duration, Instant};
+
+    use super::wait_for_rate;
+
+    /// A listener can report rates other than the target before it reports the new one, e.g. a
+    /// device stepping through rates. The whole timeout must remain available across those.
+    #[test]
+    fn wait_for_rate_honours_the_full_timeout_across_repeated_events() {
+        const TIMEOUT: Duration = Duration::from_millis(50);
+
+        let (sender, receiver) = channel::<f64>();
+        let feeder = std::thread::spawn(move || {
+            while sender.send(44_100.0).is_ok() {
+                std::thread::sleep(Duration::from_millis(1));
+            }
+        });
+
+        let start = Instant::now();
+        assert!(wait_for_rate(&receiver, 48_000, TIMEOUT).is_err());
+        let elapsed = start.elapsed();
+
+        drop(receiver);
+        let _ = feeder.join();
+
+        assert!(
+            elapsed >= TIMEOUT - Duration::from_millis(10),
+            "gave up after {elapsed:?}, well before the {TIMEOUT:?} timeout"
+        );
+    }
+
+    #[test]
+    fn wait_for_rate_returns_when_the_target_rate_is_reported() {
+        let (sender, receiver) = channel::<f64>();
+        sender.send(44_100.0).unwrap();
+        sender.send(48_000.0).unwrap();
+
+        assert!(wait_for_rate(&receiver, 48_000, Duration::from_secs(5)).is_ok());
+    }
+}