From b3ca0b2ffaa49827a23916b34a72b2c0298b8a53 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:27:02 +0200 Subject: [PATCH 01/10] fix(pulseaudio): map NoData to BackendError instead of Xrun --- src/host/pulseaudio/mod.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/host/pulseaudio/mod.rs b/src/host/pulseaudio/mod.rs index 7ddad8d9e..511207074 100644 --- a/src/host/pulseaudio/mod.rs +++ b/src/host/pulseaudio/mod.rs @@ -95,7 +95,6 @@ impl From for Error { ConnectionTerminated | Killed | Protocol | BadState | Io => { ErrorKind::StreamInvalidated } - NoData => ErrorKind::Xrun, _ => ErrorKind::BackendError, } } From 0dcc1de45b4f4020f42fcadf4ce4deb670c54655 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:27:29 +0200 Subject: [PATCH 02/10] fix(host): stop flagging try_emit_error as unused under pulseaudio-only builds --- src/host/mod.rs | 30 +++++++++++++++++++++++++++++- 1 file changed, 29 insertions(+), 1 deletion(-) diff --git a/src/host/mod.rs b/src/host/mod.rs index c5aeff56c..db8a34a77 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -196,7 +196,35 @@ pub(crate) mod error_emit; ) ), ))] -pub(crate) use error_emit::{emit_error, try_emit_error}; +pub(crate) use error_emit::emit_error; + +// Unlike `emit_error`, PulseAudio has no RT callback path and never calls this. +#[cfg(any( + target_os = "android", + target_vendor = "apple", + target_os = "windows", + all( + feature = "jack", + any( + target_os = "linux", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "netbsd", + target_os = "macos", + target_os = "windows", + ) + ), + all( + feature = "pipewire", + any( + target_os = "linux", + target_os = "dragonfly", + target_os = "freebsd", + target_os = "netbsd", + ) + ), +))] +pub(crate) use error_emit::try_emit_error; /// Convert a frame count at a given sample rate to a [`std::time::Duration`]. #[cfg(any( From 3879d82d227f346a834d165540ae46d3a6e6b5a4 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:27:38 +0200 Subject: [PATCH 03/10] fix(pipewire): don't start streams before play() is called --- src/host/pipewire/stream.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/host/pipewire/stream.rs b/src/host/pipewire/stream.rs index c1c24bd79..92692aad1 100644 --- a/src/host/pipewire/stream.rs +++ b/src/host/pipewire/stream.rs @@ -778,7 +778,7 @@ where // runs on this mainloop thread, not the separate data-loop thread RT_PROCESS creates. // That thread promotes itself to RT from the process callback, once when the negotiated // quantum is known and again whenever PipeWire renegotiates it. - let mut flags = StreamFlags::MAP_BUFFERS; + let mut flags = StreamFlags::MAP_BUFFERS | StreamFlags::INACTIVE; if connect_automatically { flags |= StreamFlags::AUTOCONNECT; } @@ -1017,7 +1017,7 @@ where // runs on this mainloop thread, not the separate data-loop thread RT_PROCESS creates. // That thread promotes itself to RT from the process callback, once when the negotiated // quantum is known and again whenever PipeWire renegotiates it. - let mut flags = StreamFlags::MAP_BUFFERS; + let mut flags = StreamFlags::MAP_BUFFERS | StreamFlags::INACTIVE; if connect_automatically { flags |= StreamFlags::AUTOCONNECT; } From a47cfc7494e6c1a00deb9a36aba33edb7bcb2847 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:27:54 +0200 Subject: [PATCH 04/10] fix(coreaudio): fill output buffers with equilibrium before the data callback --- src/host/coreaudio/ios/mod.rs | 15 ++++++++++++++- src/host/coreaudio/macos/device.rs | 5 +++++ src/host/mod.rs | 1 + 3 files changed, 20 insertions(+), 1 deletion(-) diff --git a/src/host/coreaudio/ios/mod.rs b/src/host/coreaudio/ios/mod.rs index ac1932185..dffd433ea 100644 --- a/src/host/coreaudio/ios/mod.rs +++ b/src/host/coreaudio/ios/mod.rs @@ -24,7 +24,10 @@ use self::enumerate::{ }; use super::{asbd_from_config, host_time_to_stream_instant}; use crate::{ - host::{frames_to_duration, latch::Latch, try_emit_error, ErrorCallbackArc}, + host::{ + equilibrium::fill_equilibrium, frames_to_duration, latch::Latch, try_emit_error, + ErrorCallbackArc, + }, traits::{DeviceTrait, HostTrait, StreamTrait}, BufferSize, ChannelCount, Data, DeviceDescription, DeviceDescriptionBuilder, DeviceId, Error, ErrorKind, FrameCount, InputCallbackInfo, InputStreamTimestamp, OutputCallbackInfo, @@ -624,6 +627,16 @@ where let (buffer, mut data) = unsafe { extract_audio_buffer(&args, bytes_per_channel, sample_format, false) }; + if !buffer.mData.is_null() { + let bytes = unsafe { + std::slice::from_raw_parts_mut( + buffer.mData as *mut u8, + buffer.mDataByteSize as usize, + ) + }; + fill_equilibrium(bytes, sample_format); + } + let callback = match host_time_to_stream_instant(args.time_stamp.mHostTime) { Err(err) => { error_callback(err); diff --git a/src/host/coreaudio/macos/device.rs b/src/host/coreaudio/macos/device.rs index cdd0b4807..94028d73b 100644 --- a/src/host/coreaudio/macos/device.rs +++ b/src/host/coreaudio/macos/device.rs @@ -46,6 +46,7 @@ use super::{ use crate::{ host::{ coreaudio::macos::{loopback::LoopbackDevice, StreamInner}, + equilibrium::fill_equilibrium, frames_to_duration, try_emit_error, ErrorCallbackArc, }, traits::DeviceTrait, @@ -896,6 +897,10 @@ impl Device { let data = data as *mut (); let len = data_byte_size as usize / bytes_per_channel; + + let bytes = std::slice::from_raw_parts_mut(data as *mut u8, data_byte_size as usize); + fill_equilibrium(bytes, sample_format); + let mut data = Data::from_parts(data, len, sample_format); let callback = match host_time_to_stream_instant(args.time_stamp.mHostTime) { diff --git a/src/host/mod.rs b/src/host/mod.rs index db8a34a77..85249d5f7 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -4,6 +4,7 @@ target_os = "freebsd", target_os = "netbsd", target_os = "windows", + target_vendor = "apple", ))] pub(crate) mod equilibrium; From 39221afb4e093e9d60dfc5a82db5ec942597cd47 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:28:05 +0200 Subject: [PATCH 05/10] fix(coreaudio): report xrun status on default-output streams --- src/host/coreaudio/macos/mod.rs | 127 +++++++++++++++++++++----------- 1 file changed, 85 insertions(+), 42 deletions(-) diff --git a/src/host/coreaudio/macos/mod.rs b/src/host/coreaudio/macos/mod.rs index 81be095cb..eef318e61 100644 --- a/src/host/coreaudio/macos/mod.rs +++ b/src/host/coreaudio/macos/mod.rs @@ -61,20 +61,21 @@ impl HostTrait for Host { /// Type alias for the error callback to reduce complexity type ErrorCallback = dyn FnMut(Error) + Send; -/// Spawns a dedicated thread that registers a single property listener and signals a channel on -/// each change. The listener is deregistered when the returned `Sender<()>` is dropped. -fn spawn_property_listener_thread( +/// Spawns a dedicated thread that registers a single property listener, calling `on_change` on +/// each firing. The listener is deregistered when the returned `Sender<()>` is dropped. +fn spawn_property_listener_thread( object_id: AudioObjectID, address: AudioObjectPropertyAddress, -) -> Result<(mpsc::Receiver<()>, mpsc::Sender<()>), Error> { + on_change: F, +) -> Result, Error> +where + F: FnMut() + Send + 'static, +{ let (shutdown_tx, shutdown_rx) = mpsc::channel::<()>(); - let (change_tx, change_rx) = mpsc::channel::<()>(); let (ready_tx, ready_rx) = mpsc::channel(); std::thread::spawn(move || { - let listener = AudioObjectPropertyListener::new(object_id, address, move || { - let _ = change_tx.send(()); - }); + let listener = AudioObjectPropertyListener::new(object_id, address, on_change); match listener { Ok(_l) => { let _ = ready_tx.send(Ok(())); @@ -93,7 +94,27 @@ fn spawn_property_listener_thread( ) })??; - Ok((change_rx, shutdown_tx)) + Ok(shutdown_tx) +} + +/// Registers a `kAudioDeviceProcessorOverload` listener on `device_id`, reporting `Xrun` through +/// `error_callback` on each firing (overload notifications fire on the RT thread). +fn spawn_overload_listener( + device_id: AudioDeviceID, + error_callback: &Arc>, +) -> Result, Error> { + let error_callback = error_callback.clone(); + spawn_property_listener_thread( + device_id, + AudioObjectPropertyAddress { + mSelector: kAudioDeviceProcessorOverload, + mScope: kAudioObjectPropertyScopeGlobal, + mElement: kAudioObjectPropertyElementMain, + }, + move || { + let _ = try_emit_error(&error_callback, ErrorKind::Xrun.into()); + }, + ) } /// A device monitor that can signal when the owning `Stream` handle has been returned to the @@ -247,15 +268,27 @@ impl DefaultOutputMonitor { error_callback: Arc>, latency_refresh: Option<(Arc, Scope)>, ) -> Result { - let (change_rx, shutdown_tx) = spawn_property_listener_thread( + let (change_tx, change_rx) = mpsc::channel::<()>(); + let shutdown_tx = spawn_property_listener_thread( kAudioObjectSystemObject as AudioObjectID, AudioObjectPropertyAddress { mSelector: kAudioHardwarePropertyDefaultOutputDevice, mScope: kAudioObjectPropertyScopeGlobal, mElement: kAudioObjectPropertyElementMain, }, + move || { + let _ = change_tx.send(()); + }, )?; + // The overload (xrun) listener targets a specific device, so it must be re-registered + // against whatever device is current whenever the default output reroutes. Held only to + // shut down the previous listener thread on drop when reassigned below. + let overload_error_callback = error_callback.clone(); + let mut _overload_shutdown_tx = default_output_device() + .map(|device| spawn_overload_listener(device.audio_device_id, &overload_error_callback)) + .transpose()?; + let mut latch = Latch::new(); let waiter = latch.waiter(); @@ -269,41 +302,51 @@ impl DefaultOutputMonitor { let Some(stream) = stream_weak.upgrade() else { break; }; - if default_output_device().is_none() { - if let Ok(mut inner) = stream.try_lock() { - let _ = inner.pause(); + match default_output_device() { + None => { + _overload_shutdown_tx = None; + if let Ok(mut inner) = stream.try_lock() { + let _ = inner.pause(); + } + emit_error( + &error_callback, + Error::with_message( + ErrorKind::DeviceNotAvailable, + "no default output device", + ), + ); } - emit_error( - &error_callback, - Error::with_message( - ErrorKind::DeviceNotAvailable, - "no default output device", - ), - ); - } else { - // DefaultOutput AudioUnit rerouted automatically: recompute and notify - // the buffer depth for the new device. - if let Some((frames, scope)) = &latency_refresh { - if let Ok(inner) = stream.lock() { - let depth = device::get_device_buffer_frame_size(&inner.audio_unit) - .ok() - .map_or(0, |buffer| { - buffer - + device::get_device_extra_latency_frames( - &inner.audio_unit, - *scope, - ) - }); - frames.store(depth, Ordering::Relaxed); + Some(device) => { + // DefaultOutput AudioUnit rerouted automatically: recompute and notify + // the buffer depth for the new device. + if let Some((frames, scope)) = &latency_refresh { + if let Ok(inner) = stream.lock() { + let depth = + device::get_device_buffer_frame_size(&inner.audio_unit) + .ok() + .map_or(0, |buffer| { + buffer + + device::get_device_extra_latency_frames( + &inner.audio_unit, + *scope, + ) + }); + frames.store(depth, Ordering::Relaxed); + } } + _overload_shutdown_tx = spawn_overload_listener( + device.audio_device_id, + &overload_error_callback, + ) + .ok(); + emit_error( + &error_callback, + Error::with_message( + ErrorKind::DeviceChanged, + "default output device changed", + ), + ); } - emit_error( - &error_callback, - Error::with_message( - ErrorKind::DeviceChanged, - "default output device changed", - ), - ); } } }) From 6fb1a696e5aaf4f8a4b55e0093e96d97333a0f23 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:28:14 +0200 Subject: [PATCH 06/10] fix(asio): silence the pre-callback buffer via fill_equilibrium --- src/host/asio/stream.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/host/asio/stream.rs b/src/host/asio/stream.rs index 93cc75631..e2691a618 100644 --- a/src/host/asio/stream.rs +++ b/src/host/asio/stream.rs @@ -14,6 +14,7 @@ use super::Device; use crate::{ host::{ com, + equilibrium::fill_equilibrium, error_emit::{emit_error, try_emit_error}, frames_to_duration, }, @@ -644,7 +645,7 @@ impl Device { } } - interleaved.fill(0); + fill_equilibrium(&mut interleaved, sample_format); match (sample_format, &stream_type) { (SampleFormat::I16, &sys::AsioSampleType::ASIOSTInt16LSB) => { process_output_callback::( From e03d58d756da19e9dee1bacbc5db93fd0b47d113 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:28:41 +0200 Subject: [PATCH 07/10] fix(webaudio): fix unsound Send+Sync, is_available(), and stale audio on partial writes --- Cargo.toml | 11 +- src/host/webaudio/mod.rs | 289 ++++++++++++++++++++++++++++++--------- 2 files changed, 231 insertions(+), 69 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 736eaa94b..12bb50466 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -79,7 +79,12 @@ pulseaudio = ["dep:pulseaudio", "dep:futures"] # Required for any WebAssembly audio support # Platform: WebAssembly (wasm32-unknown-unknown) # Note: This is typically enabled automatically when targeting wasm32 -wasm-bindgen = ["dep:wasm-bindgen", "dep:wasm-bindgen-futures"] +wasm-bindgen = [ + "dep:wasm-bindgen", + "dep:wasm-bindgen-futures", + "dep:futures-channel", + "dep:futures-util", +] [dependencies] dasp_sample = "0.11" @@ -174,6 +179,10 @@ objc2-avf-audio = { version = "0.3", default-features = false, features = [ [target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dependencies] wasm-bindgen = { version = "0.2", optional = true } wasm-bindgen-futures = { version = "0.4", optional = true } +futures-channel = { version = "0.3", optional = true } +futures-util = { version = "0.3", default-features = false, optional = true, features = [ + "alloc", +] } js-sys = { version = "0.3" } web-sys = { version = "0.3", features = [ "AudioContext", diff --git a/src/host/webaudio/mod.rs b/src/host/webaudio/mod.rs index 3c9119a20..d4503b7b3 100644 --- a/src/host/webaudio/mod.rs +++ b/src/host/webaudio/mod.rs @@ -6,16 +6,23 @@ extern crate js_sys; extern crate wasm_bindgen; extern crate web_sys; +#[cfg(target_feature = "atomics")] +use std::sync::atomic::AtomicU64; use std::{ fmt, ops::DerefMut, - sync::{ - atomic::{AtomicBool, Ordering}, - Arc, Mutex, RwLock, - }, + sync::{atomic::Ordering, Arc, Mutex, RwLock}, time::Duration, }; +#[cfg(not(target_feature = "atomics"))] +use std::sync::atomic::AtomicBool; + +#[cfg(target_feature = "atomics")] +use futures_channel::mpsc; +#[cfg(target_feature = "atomics")] +use futures_util::StreamExt as _; + type OutputDataCallbackArc = Arc>; use self::{ @@ -27,13 +34,19 @@ use crate::{ traits::{DeviceTrait, HostTrait, StreamTrait}, BufferSize, ChannelCount, Data, DeviceDescription, DeviceDescriptionBuilder, DeviceDirection, DeviceId, Error, ErrorKind, FrameCount, InputCallbackInfo, OutputCallbackInfo, - OutputStreamTimestamp, SampleFormat, SampleRate, StreamConfig, StreamInstant, + OutputStreamTimestamp, Sample, SampleFormat, SampleRate, StreamConfig, StreamInstant, SupportedBufferSize, SupportedStreamConfig, SupportedStreamConfigRange, }; /// Type alias for shared closure handles used in audio callbacks type ClosureHandle = Arc>>>; +#[cfg(target_feature = "atomics")] +enum Command { + Play, + Pause, +} + /// Content is false if the iterator is empty. pub struct Devices(bool); @@ -50,16 +63,32 @@ impl fmt::Display for Device { pub struct Host; pub struct Stream { + buffer_size_frames: usize, + + // Single-threaded WASM: hold JS types directly. Safe because there is only one thread. + #[cfg(not(target_feature = "atomics"))] + config: StreamConfig, + #[cfg(not(target_feature = "atomics"))] ctx: Arc, + #[cfg(not(target_feature = "atomics"))] on_ended_closures: Vec, - config: StreamConfig, - buffer_size_frames: usize, + #[cfg(not(target_feature = "atomics"))] is_started: Arc, + + // Multi-threaded WASM (+atomics): all fields are Send+Sync; JS types are owned by a + // spawn_local future on the local thread and are never stored here. + #[cfg(target_feature = "atomics")] + command_tx: mpsc::UnboundedSender, + #[cfg(target_feature = "atomics")] + current_time_bits: Arc, } -// WASM runs in a single-threaded environment, so Send and Sync are safe by design. +// Without atomics, WASM is single-threaded, so there are no thread boundaries to cross. +#[cfg(not(target_feature = "atomics"))] unsafe impl Send for Stream {} +#[cfg(not(target_feature = "atomics"))] unsafe impl Sync for Stream {} +// With atomics, all Stream fields auto-derive Send+Sync. // Compile-time assertion that Stream is Send and Sync crate::assert_stream_send!(Stream); @@ -80,9 +109,20 @@ const SUPPORTED_SAMPLE_FORMAT: SampleFormat = SampleFormat::F32; const DEFAULT_BUFFER_SIZE: usize = 2048; +// Minimum initial timer delay mandated by the HTML spec. +// https://html.spec.whatwg.org/multipage/timers-and-user-prompts.html#timers +const INITIAL_TIMEOUT_MS: i32 = 4; + impl Host { pub fn new() -> Result { - Ok(Self) + if Self::is_available() { + Ok(Self) + } else { + Err(Error::with_message( + ErrorKind::HostUnavailable, + "WebAudio is not available in this context", + )) + } } } @@ -91,8 +131,7 @@ impl HostTrait for Host { type Device = Device; fn is_available() -> bool { - // Assume this host is always available on webaudio. - true + is_webaudio_available() } fn devices(&self) -> Result { @@ -288,6 +327,8 @@ impl DeviceTrait for Device { let data_callback = crate::host::monotonic_output_callback(data_callback); let data_callback: OutputDataCallbackArc = Arc::new(Mutex::new(data_callback)); let error_callback: ErrorCallbackArc = Arc::new(Mutex::new(error_callback)); + + #[cfg(not(target_feature = "atomics"))] let is_started = Arc::new(AtomicBool::new(false)); // Create the WebAudio stream. @@ -314,7 +355,10 @@ impl DeviceTrait for Device { } destination.set_channel_count(config.channels as u32); - // SAFETY: WASM is single-threaded, so Arc is safe even though AudioContext is not Send/Sync + // SAFETY: AudioContext and Closure are not Send/Sync. In the non-atomics path WASM is + // single-threaded so there are no thread boundaries to cross. In the atomics path these + // values are moved into a spawn_local future on the same local thread; they are never + // stored in Stream (which is Send+Sync) and therefore never escape to another thread. #[allow(clippy::arc_with_non_send_sync)] let ctx = Arc::new(ctx); @@ -330,6 +374,11 @@ impl DeviceTrait for Device { .and_then(|v| v.as_f64()) .unwrap_or(0.0); + // Shared current-time counter updated on every callback invocation. + // Seeded from the live clock so now() is on the correct time base before the first callback. + #[cfg(target_feature = "atomics")] + let current_time_bits = Arc::new(AtomicU64::new(ctx.current_time().to_bits())); + // Create a set of closures / callbacks which will continuously fetch and schedule sample // playback. Starting with two workers, e.g. a front and back buffer so that audio frames // can be fetched in the background. @@ -339,6 +388,9 @@ impl DeviceTrait for Device { let ctx_handle = ctx.clone(); let time_handle = time.clone(); + #[cfg(target_feature = "atomics")] + let current_time_bits_handle = current_time_bits.clone(); + // A set of temporary buffers to be used for intermediate sample transformation steps. let mut temporary_buffer = vec![0f32; buffer_size_samples]; let mut temporary_channel_buffer = vec![0f32; buffer_size_frames]; @@ -368,7 +420,6 @@ impl DeviceTrait for Device { })?; // A self reference to this closure for passing to future audio event calls. - // SAFETY: WASM is single-threaded, so Arc is safe even though Closure is not Send/Sync #[allow(clippy::arc_with_non_send_sync)] let on_ended_closure: ClosureHandle = Arc::new(RwLock::new(None)); let on_ended_closure_handle = on_ended_closure.clone(); @@ -378,6 +429,11 @@ impl DeviceTrait for Device { .unwrap() .replace(Closure::wrap(Box::new(move || { let now = ctx_handle.current_time(); + + // Keep the shared clock up to date so Stream::now() has a fresh value. + #[cfg(target_feature = "atomics")] + current_time_bits_handle.store(now.to_bits(), Ordering::Relaxed); + let time_at_start_of_buffer = { let time_at_start_of_buffer = time_handle .read() @@ -396,6 +452,7 @@ impl DeviceTrait for Device { // Populate the sample data into an interleaved temporary buffer. { + temporary_buffer.fill(f32::EQUILIBRIUM); let len = temporary_buffer.len(); let data = temporary_buffer.as_mut_ptr() as *mut (); let mut data = unsafe { Data::from_parts(data, len, sample_format) }; @@ -568,16 +625,70 @@ impl DeviceTrait for Device { on_ended_closures.push(on_ended_closure); } - Ok(Self::Stream { - ctx, - on_ended_closures, - config, - buffer_size_frames, - is_started, - }) + #[cfg(not(target_feature = "atomics"))] + { + Ok(Self::Stream { + ctx, + on_ended_closures, + config, + buffer_size_frames, + is_started, + }) + } + + #[cfg(target_feature = "atomics")] + { + let current_time_bits_stream = current_time_bits.clone(); + let (command_tx, mut command_rx) = mpsc::unbounded::(); + wasm_bindgen_futures::spawn_local(async move { + let window = web_sys::window().unwrap(); + let mut started = false; + while let Some(cmd) = command_rx.next().await { + match cmd { + Command::Play => { + if ctx.resume().is_err() { + error_callback.lock().unwrap_or_else(|e| e.into_inner())( + Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to resume audio context", + ), + ); + } else if !started { + started = true; + schedule_initial_timeouts( + &window, + &on_ended_closures, + buffer_size_frames, + config.sample_rate, + ); + } + } + Command::Pause => { + if ctx.suspend().is_err() { + error_callback.lock().unwrap_or_else(|e| e.into_inner())( + Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to suspend audio context", + ), + ); + } + } + } + } + // Stream dropped: close the AudioContext on the main thread. + let _ = ctx.close(); + }); + Ok(Self::Stream { + command_tx, + current_time_bits: current_time_bits_stream, + buffer_size_frames, + }) + } } } +// Without atomics: AudioContext is accessible directly from Stream. +#[cfg(not(target_feature = "atomics"))] impl Stream { /// Return the [`AudioContext`](https://developer.mozilla.org/docs/Web/API/AudioContext) used /// by this stream. @@ -588,59 +699,68 @@ impl Stream { impl StreamTrait for Stream { fn play(&self) -> Result<(), Error> { - let window = web_sys::window().unwrap(); - match self.ctx.resume() { - Ok(_) => { - // Only schedule the initial timeouts once. - if self - .is_started - .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) - .is_err() - { - return Ok(()); - } - // Begin webaudio playback, initially scheduling the closures to fire on a timeout - // event. Minimum value as per spec: https://html.spec.whatwg.org/multipage/timers-and-user-prompts.html#timers - let mut offset_ms = 4; - let time_step_secs = - buffer_time_step_secs(self.buffer_size_frames, self.config.sample_rate); - let time_step_ms = ((time_step_secs * 1_000.0).ceil() as i32).max(1); - for on_ended_closure in self.on_ended_closures.iter() { - window - .set_timeout_with_callback_and_timeout_and_arguments_0( - on_ended_closure - .read() - .unwrap() - .as_ref() - .unwrap() - .as_ref() - .unchecked_ref(), - offset_ms, - ) - .unwrap(); - offset_ms += time_step_ms; + #[cfg(not(target_feature = "atomics"))] + { + let window = web_sys::window().unwrap(); + match self.ctx.resume() { + Ok(_) => { + // Only schedule the initial timeouts once. + if self + .is_started + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_err() + { + return Ok(()); + } + schedule_initial_timeouts( + &window, + &self.on_ended_closures, + self.buffer_size_frames, + self.config.sample_rate, + ); + Ok(()) } - Ok(()) + Err(_) => Err(Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to resume audio context", + )), } - Err(_) => Err(Error::with_message( - ErrorKind::DeviceNotAvailable, - "Failed to resume audio context", - )), } + #[cfg(target_feature = "atomics")] + self.command_tx.unbounded_send(Command::Play).map_err(|_| { + Error::with_message( + ErrorKind::StreamInvalidated, + "WebAudio context task stopped unexpectedly", + ) + }) } fn pause(&self) -> Result<(), Error> { - match self.ctx.suspend() { - Ok(_) => Ok(()), - Err(_) => Err(Error::with_message( - ErrorKind::DeviceNotAvailable, - "Failed to suspend audio context", - )), + #[cfg(not(target_feature = "atomics"))] + { + match self.ctx.suspend() { + Ok(_) => Ok(()), + Err(_) => Err(Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to suspend audio context", + )), + } } + #[cfg(target_feature = "atomics")] + self.command_tx.unbounded_send(Command::Pause).map_err(|_| { + Error::with_message( + ErrorKind::StreamInvalidated, + "WebAudio context task stopped unexpectedly", + ) + }) } fn now(&self) -> StreamInstant { - StreamInstant::from_secs_f64(self.ctx.current_time()) + #[cfg(not(target_feature = "atomics"))] + let t = self.ctx.current_time(); + #[cfg(target_feature = "atomics")] + let t = f64::from_bits(self.current_time_bits.load(Ordering::Relaxed)); + StreamInstant::from_secs_f64(t) } fn buffer_size(&self) -> Result { @@ -648,11 +768,15 @@ impl StreamTrait for Stream { } } +// Without atomics: close the AudioContext synchronously on drop. +#[cfg(not(target_feature = "atomics"))] impl Drop for Stream { fn drop(&mut self) { let _ = self.ctx.close(); } } +// With atomics: dropping `command_tx` closes the channel, which signals the +// spawn_local task to call ctx.close() on the main thread. impl Iterator for Devices { type Item = Device; @@ -673,24 +797,53 @@ fn default_input_device() -> Option { } fn default_output_device() -> Option { - if is_webaudio_available() { + if Host::is_available() { Some(Device) } else { None } } -// Detects whether the `AudioContext` global variable is available. +// Detects whether WebAudio is available: requires a window context (not a Worker) with an +// AudioContext constructor present. fn is_webaudio_available() -> bool { - js_sys::Reflect::get(&js_sys::global(), &JsValue::from("AudioContext")) - .unwrap() - .is_truthy() + web_sys::window() + .and_then(|w| js_sys::Reflect::get(w.as_ref(), &JsValue::from("AudioContext")).ok()) + .is_some_and(|v| v.is_truthy()) } fn buffer_time_step_secs(buffer_size_frames: usize, sample_rate: SampleRate) -> f64 { buffer_size_frames as f64 / sample_rate as f64 } +/// Stagger the initial `setTimeout` kicks for the double-buffer pair so the two closures +/// don't fire at the same instant on the first tick. +fn schedule_initial_timeouts( + window: &web_sys::Window, + on_ended_closures: &[ClosureHandle], + buffer_size_frames: usize, + sample_rate: SampleRate, +) { + let time_step_secs = buffer_time_step_secs(buffer_size_frames, sample_rate); + let time_step_ms = ((time_step_secs * 1_000.0).ceil() as i32).max(1); + let mut offset_ms = INITIAL_TIMEOUT_MS; + for on_ended_closure in on_ended_closures { + window + .set_timeout_with_callback_and_timeout_and_arguments_0( + on_ended_closure + .read() + .unwrap() + .as_ref() + .unwrap() + .as_ref() + .unchecked_ref(), + offset_ms, + ) + .unwrap(); + offset_ms += time_step_ms; + } +} + #[cfg(target_feature = "atomics")] #[wasm_bindgen] extern "C" { From 7e982bf7e20fd309247404159cbbd74f6c9846f9 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:29:01 +0200 Subject: [PATCH 08/10] fix(audioworklet): fix cross-thread Stream ops and stale audio on partial writes --- Cargo.toml | 2 + src/host/audioworklet/mod.rs | 143 ++++++++++++++++++++++++----------- 2 files changed, 99 insertions(+), 46 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 12bb50466..2f7f32d87 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,6 +41,8 @@ asio = [ # Requires: Build with atomics support and Cross-Origin headers for SharedArrayBuffer # Platform: WebAssembly (wasm32-unknown-unknown) audioworklet = [ + "dep:futures-channel", + "dep:futures-util", "wasm-bindgen", "web-sys/Blob", "web-sys/BlobPropertyBag", diff --git a/src/host/audioworklet/mod.rs b/src/host/audioworklet/mod.rs index a6173f16e..26e2327f7 100644 --- a/src/host/audioworklet/mod.rs +++ b/src/host/audioworklet/mod.rs @@ -12,6 +12,8 @@ use std::{ time::Duration, }; +use futures_channel::mpsc; +use futures_util::StreamExt as _; use js_sys::wasm_bindgen; use wasm_bindgen::prelude::*; @@ -20,7 +22,7 @@ use crate::{ traits::{DeviceTrait, HostTrait, StreamTrait}, BufferSize, ChannelCount, Data, DeviceDescription, DeviceDescriptionBuilder, DeviceDirection, DeviceId, Error, ErrorKind, FrameCount, InputCallbackInfo, OutputCallbackInfo, - OutputStreamTimestamp, SampleFormat, SampleRate, StreamConfig, StreamInstant, + OutputStreamTimestamp, Sample, SampleFormat, SampleRate, StreamConfig, StreamInstant, SupportedBufferSize, SupportedStreamConfig, SupportedStreamConfigRange, }; @@ -42,10 +44,21 @@ impl fmt::Display for Device { pub struct Host; +// `Stream`'s fields are all Send+Sync (a channel sender plus atomics). +// Any JS-backed resource (e.g. `web_sys::Window`, `Closure`) must live as a local +// inside the `spawn_local` task in `build_*_stream_raw`. pub struct Stream { - audio_context: web_sys::AudioContext, + command_tx: mpsc::UnboundedSender, + current_time_bits: Arc, buffer_size_frames: Arc, - _latency_poller: Option, +} + +crate::assert_stream_send!(Stream); +crate::assert_stream_sync!(Stream); + +enum Command { + Play, + Pause, } /// How often the main thread re-reads `outputLatency` to publish it to the worklet. @@ -353,6 +366,13 @@ impl DeviceTrait for Device { }); let buffer_size_frames = Arc::new(AtomicU64::new(initial_quantum)); let buffer_size_frames_cb = buffer_size_frames.clone(); + + let current_time_bits = Arc::new(AtomicU64::new(audio_context.current_time().to_bits())); + let current_time_bits_cb = current_time_bits.clone(); + let current_time_bits_init = current_time_bits.clone(); + + let (command_tx, mut command_rx) = mpsc::unbounded::(); + // The worklet realm cannot read AudioContext properties, so share the value via an atomic. let latency_nanos = Arc::new(AtomicU64::new(total_latency_nanos(&audio_context))); let latency_nanos_cb = latency_nanos.clone(); @@ -377,6 +397,7 @@ impl DeviceTrait for Device { &WasmAudioProcessor::new(Box::new( move |interleaved_data, frame_size, sample_rate, now| { buffer_size_frames_cb.store(frame_size as u64, Ordering::Relaxed); + current_time_bits_cb.store(now.to_bits(), Ordering::Relaxed); let data = interleaved_data.as_mut_ptr() as *mut (); let mut data = unsafe { Data::from_parts(data, interleaved_data.len(), sample_format) @@ -412,35 +433,69 @@ impl DeviceTrait for Device { error_callback(Error::with_message( ErrorKind::UnsupportedOperation, message, - )) + )); + + // Close AudioContext and exit; dropping command_rx closes the channel, + // so subsequent play()/pause() calls return HostUnavailable. + let _ = audio_context.close(); + return; } - }); - // outputLatency can change at runtime (e.g. an output-device switch) but is only readable - // on the main thread, so poll it here and publish it to the worklet via the shared atomic. - let latency_poller = web_sys::window().and_then(|window| { - let poll_ctx = audio_context.clone(); - let poll_latency = latency_nanos.clone(); - let closure = Closure::::new(move || { - poll_latency.store(total_latency_nanos(&poll_ctx), Ordering::Relaxed); + current_time_bits_init.store(audio_context.current_time().to_bits(), Ordering::Relaxed); + + // outputLatency can change at runtime (e.g. an output-device switch) but is only + // readable on the main thread, so poll it here and publish it to the worklet via the + // shared atomic. + let _latency_poller = web_sys::window().and_then(|window| { + let poll_ctx = audio_context.clone(); + let poll_latency = latency_nanos.clone(); + let closure = Closure::::new(move || { + poll_latency.store(total_latency_nanos(&poll_ctx), Ordering::Relaxed); + }); + window + .set_interval_with_callback_and_timeout_and_arguments_0( + closure.as_ref().unchecked_ref(), + LATENCY_POLL_INTERVAL.as_millis() as i32, + ) + .ok() + .map(|interval_id| LatencyPoller { + window, + interval_id, + _closure: closure, + }) }); - window - .set_interval_with_callback_and_timeout_and_arguments_0( - closure.as_ref().unchecked_ref(), - LATENCY_POLL_INTERVAL.as_millis() as i32, - ) - .ok() - .map(|interval_id| LatencyPoller { - window, - interval_id, - _closure: closure, - }) + + // Process play/pause commands from any thread until Stream is dropped. + // Dropping Stream closes command_tx, which terminates this loop. + while let Some(cmd) = command_rx.next().await { + match cmd { + Command::Play => { + if audio_context.resume().is_err() { + error_callback(Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to resume audio context", + )); + } + } + Command::Pause => { + if audio_context.suspend().is_err() { + error_callback(Error::with_message( + ErrorKind::DeviceNotAvailable, + "Failed to suspend audio context", + )); + } + } + } + } + + // Stream dropped: close the AudioContext on the main thread. + let _ = audio_context.close(); }); Ok(Self::Stream { - audio_context, + command_tx, + current_time_bits, buffer_size_frames, - _latency_poller: latency_poller, }) } } @@ -451,33 +506,27 @@ impl StreamTrait for Stream { } fn play(&self) -> Result<(), Error> { - match self.audio_context.resume() { - Ok(_) => Ok(()), - Err(_) => Err(Error::with_message( - ErrorKind::DeviceNotAvailable, - "Failed to resume audio context", - )), - } + self.command_tx.unbounded_send(Command::Play).map_err(|_| { + Error::with_message( + ErrorKind::HostUnavailable, + "audio worklet initialization failed", + ) + }) } fn pause(&self) -> Result<(), Error> { - match self.audio_context.suspend() { - Ok(_) => Ok(()), - Err(_) => Err(Error::with_message( - ErrorKind::DeviceNotAvailable, - "Failed to suspend audio context", - )), - } + self.command_tx.unbounded_send(Command::Pause).map_err(|_| { + Error::with_message( + ErrorKind::HostUnavailable, + "audio worklet initialization failed", + ) + }) } fn now(&self) -> StreamInstant { - StreamInstant::from_secs_f64(self.audio_context.current_time()) - } -} - -impl Drop for Stream { - fn drop(&mut self) { - let _ = self.audio_context.close(); + StreamInstant::from_secs_f64(f64::from_bits( + self.current_time_bits.load(Ordering::Relaxed), + )) } } @@ -533,6 +582,8 @@ impl WasmAudioProcessor { 0.0, ); + self.interleaved_buffer[..interleaved_buffer_size].fill(f32::EQUILIBRIUM); + (self.callback)( &mut self.interleaved_buffer[..interleaved_buffer_size], frame_size as u32, From 7bae17b7a81109de398ceb957a0dc8c9b3f1a2ad Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:29:13 +0200 Subject: [PATCH 09/10] fix: feature-gate the custom example in Cargo.toml instead of the example itself --- Cargo.toml | 4 ++++ examples/custom.rs | 1 - 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 2f7f32d87..9b2f6e878 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -224,6 +224,10 @@ name = "record_wav" [[example]] name = "synth_tones" +[[example]] +name = "custom" +required-features = ["custom"] + [package.metadata.docs.rs] # The docs.rs build environment does not have all the dependencies for all features features = [ diff --git a/examples/custom.rs b/examples/custom.rs index 4b09cf911..a2dcc2d09 100644 --- a/examples/custom.rs +++ b/examples/custom.rs @@ -15,7 +15,6 @@ use cpal::{ SupportedBufferSize, SupportedStreamConfig, SupportedStreamConfigRange, }; -#[allow(dead_code)] #[derive(Clone)] // Clone, Send+Sync are required struct MyHost; From bff7a2691e439ff198abad792fb88851e96066e2 Mon Sep 17 00:00:00 2001 From: Roderick van Domburg Date: Mon, 13 Jul 2026 22:29:23 +0200 Subject: [PATCH 10/10] docs: add v0.18.2 changelog entries and document the pre-filled-silence guarantee --- CHANGELOG.md | 9 +++++++++ src/traits.rs | 6 ++++-- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 22c93b41c..46f35cd69 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,13 +25,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Timestamps now stay monotonic across device and graph changes. - **ALSA**: A nonzero but sub-millisecond stream timeout is no longer treated as a non-blocking poll. +- **AudioWorklet**: Fix `Stream` operations to work when called from any thread. +- **AudioWorklet**: Fix stale audio output when a data callback wrote a partial buffer. +- **CoreAudio**: Default-output streams now report xrun status. +- **CoreAudio**: Fix stale audio output when a data callback wrote a partial buffer. - **JACK**: Streams with more channels than physical system ports no longer fail to build. - **JACK**: Channel enumeration is no longer capped at the physical system port count. +- **PipeWire**: Fix streams starting audio before `play()` is called. - **PipeWire**: Streams for a specific device no longer auto-reroute if it disappears. +- **PulseAudio**: `NoData` errors are no longer misreported as buffer xruns. - **visionOS**: The CoreAudio backend now builds. - **WASAPI**: Default device changes no longer report `DeviceChanged`, which wrongly implied the stream had rerouted automatically. - **WASAPI**: Reported buffer sizes are no longer off by one frame. - **WASAPI**: `Stream::drop`, `play`, and `pause` no longer panic when the device is lost. +- **WebAudio**: Fix stale audio output when a data callback wrote a partial buffer. +- **WebAudio**: Fix unsound `Send + Sync` on `Stream` when compiled with `+atomics`. +- **WebAudio**: Fix `Host::is_available()` always returning `true`, even in non-window contexts. ## [0.18.1] - 2026-06-07 diff --git a/src/traits.rs b/src/traits.rs index 5ef3b554e..cc0b91124 100644 --- a/src/traits.rs +++ b/src/traits.rs @@ -286,7 +286,8 @@ pub trait DeviceTrait: PartialEq + Eq + Hash + Debug + Display { /// * `config` - The stream configuration including sample rate, channels, and buffer size. /// * `data_callback` - Called periodically to fill the output buffer. The callback receives /// a mutable slice of samples in the format `T` to be filled with audio data, along with - /// timing information. + /// timing information. The slice is pre-filled with silence, so a callback that writes + /// fewer samples than the slice holds leaves the remainder silent rather than stale. /// * `error_callback` - Called when a stream error occurs (e.g., device disconnected). /// * `timeout` - Time to wait for the backend to initialize the stream. `None` waits /// indefinitely; `Some(duration)` limits how long to wait. Note: not all backends honor @@ -390,7 +391,8 @@ pub trait DeviceTrait: PartialEq + Eq + Hash + Debug + Display { /// * `config` - The stream configuration including sample rate, channels, and buffer size. /// * `sample_format` - The sample format of the audio data. /// * `data_callback` - Called periodically to fill the output buffer with audio data as - /// a mutable [`Data`] buffer. + /// a mutable [`Data`] buffer. The buffer is pre-filled with silence, so a callback that + /// writes fewer samples than the buffer holds leaves the remainder silent rather than stale. /// * `error_callback` - Called when a stream error occurs (e.g., device disconnected). /// * `timeout` - Time to wait for the backend to initialize the stream. `None` waits /// indefinitely; `Some(duration)` limits how long to wait. Note: not all backends honor